<p dir="auto">Hi all,</p>
<p dir="auto">I seeing sometimes failed Kafka indexing tasks, the task log look quite empty (~180 lines) - ends with this -</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="2018-02-05T03:32:28,045 INFO [main] io.druid.guice.JsonConfigurator - Loaded class[class io.druid.query.search.search.SearchQueryConfig] from props[druid.query.search.] as [io.druid.query.search.search.SearchQueryConfig@7c3e4b1a]
2018-02-05T03:32:28,048 INFO [main] io.druid.guice.JsonConfigurator - Loaded class[class io.druid.query.metadata.SegmentMetadataQueryConfig] from props[druid.query.segmentMetadata.] as [io.druid.query.metadata.SegmentMetadataQueryConfig@41a374be]
2018-02-05T03:32:28,052 INFO [main] io.druid.guice.JsonConfigurator - Loaded class[class io.druid.query.groupby.GroupByQueryConfig] from props[druid.query.groupBy.] as [io.druid.query.groupby.GroupByQueryConfig@5f96f6a2]
2018-02-05T03:32:28,071 INFO [main] io.druid.offheap.OffheapBufferGenerator - Allocating new intermediate processing buffer[0] of size[536,870,912]
2018-02-05T03:32:41,994 INFO [main] io.druid.offheap.OffheapBufferGenerator - Allocating new intermediate processing buffer[1] of size[536,870,912]
2018-02-05T03:32:56,979 INFO [main] io.druid.offheap.OffheapBufferGenerator - Allocating new result merging buffer[0] of size[536,870,912]
2018-02-05T03:33:12,481 INFO [main] io.druid.offheap.OffheapBufferGenerator - Allocating new result merging buffer[1] of size[536,870,912]"><pre class="notranslate"><code class="notranslate">2018-02-05T03:32:28,045 INFO [main] io.druid.guice.JsonConfigurator - Loaded class[class io.druid.query.search.search.SearchQueryConfig] from props[druid.query.search.] as [io.druid.query.search.search.SearchQueryConfig@7c3e4b1a]
2018-02-05T03:32:28,048 INFO [main] io.druid.guice.JsonConfigurator - Loaded class[class io.druid.query.metadata.SegmentMetadataQueryConfig] from props[druid.query.segmentMetadata.] as [io.druid.query.metadata.SegmentMetadataQueryConfig@41a374be]
2018-02-05T03:32:28,052 INFO [main] io.druid.guice.JsonConfigurator - Loaded class[class io.druid.query.groupby.GroupByQueryConfig] from props[druid.query.groupBy.] as [io.druid.query.groupby.GroupByQueryConfig@5f96f6a2]
2018-02-05T03:32:28,071 INFO [main] io.druid.offheap.OffheapBufferGenerator - Allocating new intermediate processing buffer[0] of size[536,870,912]
2018-02-05T03:32:41,994 INFO [main] io.druid.offheap.OffheapBufferGenerator - Allocating new intermediate processing buffer[1] of size[536,870,912]
2018-02-05T03:32:56,979 INFO [main] io.druid.offheap.OffheapBufferGenerator - Allocating new result merging buffer[0] of size[536,870,912]
2018-02-05T03:33:12,481 INFO [main] io.druid.offheap.OffheapBufferGenerator - Allocating new result merging buffer[1] of size[536,870,912]
</code></pre></div>
<p dir="auto">Looking at the middle manager, I see this exception -</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="2018-02-05T03:32:20,956 INFO [WorkerTaskMonitor] io.druid.indexing.worker.WorkerTaskMonitor - Submitting runnable for task[index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa]
2018-02-05T03:32:20,960 INFO [WorkerTaskMonitor] io.druid.indexing.worker.WorkerTaskMonitor - Affirmative. Running task [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa]
2018-02-05T03:32:20,963 INFO [forking-task-runner-3] io.druid.indexing.overlord.ForkingTaskRunner - Running command: java -cp /opt/kava/conf/druid/_common:/opt/kava/conf/druid/middleManager:lib/jetty-continuation-9.3.16.v20170120.jar:lib/druid-aws-common-0.10.0.jar:lib/java-xmlbuilder-1.1.jar:lib/...
2018-02-05T03:32:20,965 INFO [forking-task-runner-3] io.druid.indexing.overlord.TaskRunnerUtils - Task [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa] location changed to [TaskLocation{host='ip-x-y-z-w.node.us-west-2.consul', port=8102}].
2018-02-05T03:32:20,965 INFO [forking-task-runner-3] io.druid.indexing.overlord.TaskRunnerUtils - Task [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa] status changed to [RUNNING].
2018-02-05T03:32:20,965 INFO [forking-task-runner-3] io.druid.indexing.overlord.ForkingTaskRunner - Logging task index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa output to: var/druid/task/index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa/log
2018-02-05T03:32:20,965 INFO [WorkerTaskMonitor] io.druid.indexing.worker.WorkerTaskMonitor - Updating task [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa] announcement with location [TaskLocation{host='ip-x-y-z-w.node.us-west-2.consul', port=8102}]
2018-02-05T03:33:14,778 INFO [HttpPostEmitter-1-0] com.metamx.http.client.pool.ChannelResourceFactory - Generating: http://json-push.service.us-west-2.consul:7000
2018-02-05T03:33:27,265 INFO [qtp81907268-47] io.druid.indexing.overlord.ForkingTaskRunner - Killing process for task: index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa
2018-02-05T03:33:27,286 INFO [qtp81907268-44] io.druid.indexing.overlord.ForkingTaskRunner - Killing process for task: index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa
2018-02-05T03:33:27,332 INFO [qtp81907268-57] io.druid.indexing.overlord.ForkingTaskRunner - Killing process for task: index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa
2018-02-05T03:33:27,466 INFO [forking-task-runner-3] io.druid.storage.s3.S3TaskLogs - Pushing task log var/druid/task/index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa/log to: druid/indexing-logs/index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa/log
2018-02-05T03:33:27,567 INFO [forking-task-runner-3] io.druid.indexing.overlord.ForkingTaskRunner - Exception caught during execution
java.io.IOException: Stream closed
        at java.io.BufferedInputStream.getBufIfOpen(BufferedInputStream.java:170) ~[?:1.8.0_131]
        at java.io.BufferedInputStream.read1(BufferedInputStream.java:291) ~[?:1.8.0_131]
        at java.io.BufferedInputStream.read(BufferedInputStream.java:345) ~[?:1.8.0_131]
        at java.io.FilterInputStream.read(FilterInputStream.java:107) ~[?:1.8.0_131]
        at com.google.common.io.ByteStreams.copy(ByteStreams.java:175) ~[guava-16.0.1.jar:?]
        at io.druid.indexing.overlord.ForkingTaskRunner$1.call(ForkingTaskRunner.java:438) [druid-indexing-service-0.10.0.jar:0.10.0]
        at io.druid.indexing.overlord.ForkingTaskRunner$1.call(ForkingTaskRunner.java:220) [druid-indexing-service-0.10.0.jar:0.10.0]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_131]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_131]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_131]
        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131]
2018-02-05T03:33:27,568 INFO [forking-task-runner-3] io.druid.indexing.overlord.ForkingTaskRunner - Removing task directory: var/druid/task/index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa
2018-02-05T03:33:27,572 INFO [WorkerTaskMonitor] io.druid.indexing.worker.WorkerTaskMonitor - Job's finished. Completed [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa] with status [FAILED]"><pre class="notranslate"><code class="notranslate">2018-02-05T03:32:20,956 INFO [WorkerTaskMonitor] io.druid.indexing.worker.WorkerTaskMonitor - Submitting runnable for task[index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa]
2018-02-05T03:32:20,960 INFO [WorkerTaskMonitor] io.druid.indexing.worker.WorkerTaskMonitor - Affirmative. Running task [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa]
2018-02-05T03:32:20,963 INFO [forking-task-runner-3] io.druid.indexing.overlord.ForkingTaskRunner - Running command: java -cp /opt/kava/conf/druid/_common:/opt/kava/conf/druid/middleManager:lib/jetty-continuation-9.3.16.v20170120.jar:lib/druid-aws-common-0.10.0.jar:lib/java-xmlbuilder-1.1.jar:lib/...
2018-02-05T03:32:20,965 INFO [forking-task-runner-3] io.druid.indexing.overlord.TaskRunnerUtils - Task [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa] location changed to [TaskLocation{host='ip-x-y-z-w.node.us-west-2.consul', port=8102}].
2018-02-05T03:32:20,965 INFO [forking-task-runner-3] io.druid.indexing.overlord.TaskRunnerUtils - Task [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa] status changed to [RUNNING].
2018-02-05T03:32:20,965 INFO [forking-task-runner-3] io.druid.indexing.overlord.ForkingTaskRunner - Logging task index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa output to: var/druid/task/index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa/log
2018-02-05T03:32:20,965 INFO [WorkerTaskMonitor] io.druid.indexing.worker.WorkerTaskMonitor - Updating task [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa] announcement with location [TaskLocation{host='ip-x-y-z-w.node.us-west-2.consul', port=8102}]
2018-02-05T03:33:14,778 INFO [HttpPostEmitter-1-0] com.metamx.http.client.pool.ChannelResourceFactory - Generating: http://json-push.service.us-west-2.consul:7000
2018-02-05T03:33:27,265 INFO [qtp81907268-47] io.druid.indexing.overlord.ForkingTaskRunner - Killing process for task: index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa
2018-02-05T03:33:27,286 INFO [qtp81907268-44] io.druid.indexing.overlord.ForkingTaskRunner - Killing process for task: index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa
2018-02-05T03:33:27,332 INFO [qtp81907268-57] io.druid.indexing.overlord.ForkingTaskRunner - Killing process for task: index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa
2018-02-05T03:33:27,466 INFO [forking-task-runner-3] io.druid.storage.s3.S3TaskLogs - Pushing task log var/druid/task/index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa/log to: druid/indexing-logs/index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa/log
2018-02-05T03:33:27,567 INFO [forking-task-runner-3] io.druid.indexing.overlord.ForkingTaskRunner - Exception caught during execution
java.io.IOException: Stream closed
        at java.io.BufferedInputStream.getBufIfOpen(BufferedInputStream.java:170) ~[?:1.8.0_131]
        at java.io.BufferedInputStream.read1(BufferedInputStream.java:291) ~[?:1.8.0_131]
        at java.io.BufferedInputStream.read(BufferedInputStream.java:345) ~[?:1.8.0_131]
        at java.io.FilterInputStream.read(FilterInputStream.java:107) ~[?:1.8.0_131]
        at com.google.common.io.ByteStreams.copy(ByteStreams.java:175) ~[guava-16.0.1.jar:?]
        at io.druid.indexing.overlord.ForkingTaskRunner$1.call(ForkingTaskRunner.java:438) [druid-indexing-service-0.10.0.jar:0.10.0]
        at io.druid.indexing.overlord.ForkingTaskRunner$1.call(ForkingTaskRunner.java:220) [druid-indexing-service-0.10.0.jar:0.10.0]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_131]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_131]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_131]
        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131]
2018-02-05T03:33:27,568 INFO [forking-task-runner-3] io.druid.indexing.overlord.ForkingTaskRunner - Removing task directory: var/druid/task/index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa
2018-02-05T03:33:27,572 INFO [WorkerTaskMonitor] io.druid.indexing.worker.WorkerTaskMonitor - Job's finished. Completed [index_kafka_player-events-realtime_f99c5d8777e2890_igkdbboa] with status [FAILED]
</code></pre></div>
<p dir="auto">I saw this exception was reported in <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="158126052" data-permission-text="Title is private" data-url="https://github.com/apache/druid/issues/3054" data-hovercard-type="issue" data-hovercard-url="/apache/druid/issues/3054/hovercard" href="https://github.com/apache/druid/issues/3054">#3054</a>, but this seems to be different, as here it's sporadic. I'm mainly concerned about whether I may be losing events when it happens. If the code just spawns another task that continues from the same position, then I guess it can be ignored...</p>
<p dir="auto">Thanks!</p>
<p dir="auto">Eran</p>
<p dir="auto"><strong>Steps to reproduce:</strong></p>
<ol dir="auto">
<li>Create a <a href="http://druid.io/docs/0.9.1.1/development/extensions-core/kafka-ingestion.html" rel="nofollow">kafka indexer</a></li>
<li>Create a <a href="http://druid.io/docs/0.9.1.1/development/extensions-core/kafka-extraction-namespace.html" rel="nofollow">kafka lookup</a></li>
<li>Wait for a kafka index task to start</li>
<li>Emulate a <code class="notranslate">/stop</code> request sent from supervisor to worker, e.g. via</li>
</ol>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="http post localhost:8100/druid/worker/v1/chat/index_kafka_Example_c2f17f3de1b02c1_fmcaakbh/stop"><pre class="notranslate"><code class="notranslate">http post localhost:8100/druid/worker/v1/chat/index_kafka_Example_c2f17f3de1b02c1_fmcaakbh/stop
</code></pre></div>
<p dir="auto"><strong>Expected behavior:</strong><br>
Worker successfully stops</p>
<p dir="auto"><strong>Actual behavior:</strong><br>
Worker process remains lingering until middle manager forcefully kills the process after timeout</p>
<hr>
<p dir="auto">The reason for this is that when <code class="notranslate">KafkaLookupExtractorFactory</code> invokes <code class="notranslate">consumerConnector.shutdown()</code> upon receiving the stop command, it interrupts the worker thread, and thus the shutdown does not complete properly:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2072) ~[?:1.7.0_101]
    at java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1468) ~[?:1.7.0_101]
    at kafka.utils.KafkaScheduler.shutdown(KafkaScheduler.scala:88) ~[kafka_2.10-0.8.2.1.jar:?]
    at kafka.consumer.ZookeeperConsumerConnector.liftedTree1$1(ZookeeperConsumerConnector.scala:195) [kafka_2.10-0.8.2.1.jar:?]
    at kafka.consumer.ZookeeperConsumerConnector.shutdown(ZookeeperConsumerConnector.scala:193) [kafka_2.10-0.8.2.1.jar:?]
    at kafka.javaapi.consumer.ZookeeperConsumerConnector.shutdown(ZookeeperConsumerConnector.scala:119) [kafka_2.10-0.8.2.1.jar:?]
    at io.druid.query.lookup.KafkaLookupExtractorFactory$2.run(KafkaLookupExtractorFactory.java:229) [druid-kafka-extraction-namespace-0.9.2-SNAPSHOT.jar:0.9.1.1]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) [?:1.7.0_101]
    at java.util.concurrent.FutureTask.run(FutureTask.java:262) [?:1.7.0_101]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [?:1.7.0_101]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [?:1.7.0_101]
    at java.lang.Thread.run(Thread.java:745) [?:1.7.0_101]"><pre class="notranslate"><code class="notranslate">java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2072) ~[?:1.7.0_101]
    at java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1468) ~[?:1.7.0_101]
    at kafka.utils.KafkaScheduler.shutdown(KafkaScheduler.scala:88) ~[kafka_2.10-0.8.2.1.jar:?]
    at kafka.consumer.ZookeeperConsumerConnector.liftedTree1$1(ZookeeperConsumerConnector.scala:195) [kafka_2.10-0.8.2.1.jar:?]
    at kafka.consumer.ZookeeperConsumerConnector.shutdown(ZookeeperConsumerConnector.scala:193) [kafka_2.10-0.8.2.1.jar:?]
    at kafka.javaapi.consumer.ZookeeperConsumerConnector.shutdown(ZookeeperConsumerConnector.scala:119) [kafka_2.10-0.8.2.1.jar:?]
    at io.druid.query.lookup.KafkaLookupExtractorFactory$2.run(KafkaLookupExtractorFactory.java:229) [druid-kafka-extraction-namespace-0.9.2-SNAPSHOT.jar:0.9.1.1]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) [?:1.7.0_101]
    at java.util.concurrent.FutureTask.run(FutureTask.java:262) [?:1.7.0_101]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [?:1.7.0_101]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [?:1.7.0_101]
    at java.lang.Thread.run(Thread.java:745) [?:1.7.0_101]
</code></pre></div>
<p dir="auto">This leaves a lingering <code class="notranslate">ConsumerFetcherThread</code> behind, and so the process does not terminate.</p>