Kafka Indexing Service stops managing tasks and ingesting data when active task ingests 0 events before completionTimeout

[acdn-ekeddy]

Occurring in Druid 0.12.0.
An NPE is thrown in the KafkaSupervisor when an active task reaches the completion timeout and it has not ingested any events.
The KafkaSupervisor keeps looping and hitting this exception. No new tasks are started after this exception so when new data appears in the topic, it doesn't get consumed.
Resetting the supervisor gets us out of this state.

2018-04-21T15:56:26,807 ERROR [KafkaSupervisor-npav-ts-metrics] io.druid.indexing.kafka.supervisor.KafkaSupervisor - No task in [[index_kafka_npav-ts-metrics_33555930f94bdd9_dlidblmf]] succeeded before the completion timeout elapsed [PT600S]!: {class=io.druid.indexing.kafka.supervisor.KafkaSupervisor}
2018-04-21T15:54:26,836 ERROR [KafkaSupervisor-npav-ts-metrics] io.druid.indexing.kafka.supervisor.KafkaSupervisor - KafkaSupervisor[npav-ts-metrics] failed to handle notice: {class=io.druid.indexing.kafka.supervisor.KafkaSupervisor, exceptionType=class java.lang.NullPointerException, exceptionMessage=null, noticeClass=RunNotice}
java.lang.NullPointerException
	at io.druid.indexing.kafka.supervisor.KafkaSupervisor.generateSequenceName(KafkaSupervisor.java:873) ~[druid-kafka-indexing-service-0.12.0.jar:0.12.0]
	at io.druid.indexing.kafka.supervisor.KafkaSupervisor.checkPendingCompletionTasks(KafkaSupervisor.java:1615) ~[druid-kafka-indexing-service-0.12.0.jar:0.12.0]
	at io.druid.indexing.kafka.supervisor.KafkaSupervisor.runInternal(KafkaSupervisor.java:823) ~[druid-kafka-indexing-service-0.12.0.jar:0.12.0]
	at io.druid.indexing.kafka.supervisor.KafkaSupervisor$RunNotice.handle(KafkaSupervisor.java:574) ~[druid-kafka-indexing-service-0.12.0.jar:0.12.0]
	at io.druid.indexing.kafka.supervisor.KafkaSupervisor$2.run(KafkaSupervisor.java:366) [druid-kafka-indexing-service-0.12.0.jar:0.12.0]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_66-internal]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_66-internal]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_66-internal]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_66-internal]
	at java.lang.Thread.run(Thread.java:745) [?:1.8.0_66-internal]

In my setup I installed Druid 0.12.0 using http://static.druid.io/artifacts/releases/druid-0.12.0-bin.tar.gz. I created a supervisor task using the following tuning and io config:

"tuningConfig": {
    "type": "kafka",
    "reportParseExceptions": false,
    "maxRowsPerSegment": 5000000,
    "intermediatePersistPeriod": "PT5M",
    "resetOffsetAutomatically": true,
    "handoffConditionTimeout": 300000
  },
  "ioConfig": {
    "topic": "npav-ts-metrics",
    "consumerProperties": {
      "bootstrap.servers": "kafka:9092"
    },
    "useEarliestOffset":true,
    "replicas": 1,
    "taskCount": 1,
    "taskDuration": "PT10M",
    "earlyMessageRejectionPeriod": "PT1H",
    "completionTimeout":"PT10M"
  }

I send some events through and then stop sending events. The first task task completes successfully since it processed events. A 2nd task is started and continues running past the 10-minute taskDuration period. A 3rd task is started, as expected. Eventually, around the 20 minutes after the 2nd task started, the overlord hits the NPE.
I begin sending new events but nothing gets processed.

[acdn-ekeddy]

If I reduce the completionTimeout to PT2M, there is no NPE. The task, however, is always marked failed.

[jon-wei]

Going to merge this with issue #5656, I think these are the same issue