Issue 1:
Hi all,

I seeing sometimes failed Kafka indexing tasks, the task log look quite empty (~180 lines) - ends with this -

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]

Looking at the middle manager, I see this exception -

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]

I saw this exception was reported in #3054, 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...

Thanks!

Eran

Issue 2:
Steps to reproduce:

Create a kafka indexer

Create a kafka lookup

Wait for a kafka index task to start

Emulate a /stop request sent from supervisor to worker, e.g. via

http post localhost:8100/druid/worker/v1/chat/index_kafka_Example_c2f17f3de1b02c1_fmcaakbh/stop

Expected behavior:

Worker successfully stops

Actual behavior:

Worker process remains lingering until middle manager forcefully kills the process after timeout

The reason for this is that when KafkaLookupExtractorFactory invokes consumerConnector.shutdown() upon receiving the stop command, it interrupts the worker thread, and thus the shutdown does not complete properly:

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]

This leaves a lingering ConsumerFetcherThread behind, and so the process does not terminate.