[KAFKA-2944] NullPointerException in KafkaConfigStorage when config storage starts right before shutdown request

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.9.0.0
• Fix Version/s: 0.10.0.0
• Component/s: connect
• Assignee: Gwen Shapira
• Reporter: Ewen Cheslack-Postava
• Reviewer: Ewen Cheslack-Postava
• Created: 03/Dec/15 16:58
• Updated: 12/Jan/17 02:05
• Resolved: 15/Jun/16 15:53

Description

Relevant log where you can see a config update starting, then the request to shutdown happens and we end up with a NullPointerException:

[2015-12-03 09:12:55,712] DEBUG Change in connector task count from 2 to 3, writing updated task configurations (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2015-12-03 09:12:56,224] INFO Kafka Connect stopping (org.apache.kafka.connect.runtime.Connect)
[2015-12-03 09:12:56,224] INFO Stopping REST server (org.apache.kafka.connect.runtime.rest.RestServer)
[2015-12-03 09:12:56,227] INFO Stopped ServerConnector@10cb550e

Unknown macro: {HTTP/1.1}

Unknown macro: {0.0.0.0}

(org.eclipse.jetty.server.ServerConnector)
[2015-12-03 09:12:56,234] INFO Stopped o.e.j.s.ServletContextHandler@3f8a24d5

Unknown macro: {/,null,UNAVAILABLE}

(org.eclipse.jetty.server.handler.ContextHandler)
[2015-12-03 09:12:56,235] INFO REST server stopped (org.apache.kafka.connect.runtime.rest.RestServer)
[2015-12-03 09:12:56,235] INFO Herder stopping (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2015-12-03 09:12:58,209] ERROR Unexpected exception in KafkaBasedLog's work thread (org.apache.kafka.connect.util.KafkaBasedLog)
java.lang.NullPointerException
at org.apache.kafka.connect.storage.KafkaConfigStorage.completeTaskIdSet(KafkaConfigStorage.java:558)
at org.apache.kafka.connect.storage.KafkaConfigStorage.access$1200(KafkaConfigStorage.java:143)
at org.apache.kafka.connect.storage.KafkaConfigStorage$1.onCompletion(KafkaConfigStorage.java:476)
at org.apache.kafka.connect.storage.KafkaConfigStorage$1.onCompletion(KafkaConfigStorage.java:372)
at org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:235)
at org.apache.kafka.connect.util.KafkaBasedLog.readToLogEnd(KafkaBasedLog.java:275)
at org.apache.kafka.connect.util.KafkaBasedLog.access$300(KafkaBasedLog.java:70)
at org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:307)
[2015-12-03 09:13:26,704] ERROR Failed to write root configuration to Kafka: (org.apache.kafka.connect.storage.KafkaConfigStorage)
java.util.concurrent.TimeoutException: Timed out waiting for future
at org.apache.kafka.connect.util.ConvertingFutureCallback.get(ConvertingFutureCallback.java:74)
at org.apache.kafka.connect.storage.KafkaConfigStorage.putTaskConfigs(KafkaConfigStorage.java:352)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnector(DistributedHerder.java:737)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnectorTasksWithRetry(DistributedHerder.java:677)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:673)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startWork(DistributedHerder.java:640)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.handleRebalanceCompleted(DistributedHerder.java:598)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.tick(DistributedHerder.java:184)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:159)
at java.lang.Thread.run(Thread.java:745)
[2015-12-03 09:13:26,704] ERROR Failed to reconfigure connector's tasks, retrying after backoff: (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
org.apache.kafka.connect.errors.ConnectException: Error writing root configuration to Kafka
at org.apache.kafka.connect.storage.KafkaConfigStorage.putTaskConfigs(KafkaConfigStorage.java:355)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnector(DistributedHerder.java:737)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnectorTasksWithRetry(DistributedHerder.java:677)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:673)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startWork(DistributedHerder.java:640)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.handleRebalanceCompleted(DistributedHerder.java:598)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.tick(DistributedHerder.java:184)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:159)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.util.concurrent.TimeoutException: Timed out waiting for future
at org.apache.kafka.connect.util.ConvertingFutureCallback.get(ConvertingFutureCallback.java:74)
at org.apache.kafka.connect.storage.KafkaConfigStorage.putTaskConfigs(KafkaConfigStorage.java:352)
... 8 more

I'm not certain that the issue is specifically due to shutting down (the KafkaConfigStorage.stop() hasn't been invoked yet when this occurs, so the underlying KafkaBasedLog is still running, although shutdown of the entire process has started), but this has only shown up during shutdown so far.

Issue Links

• is duplicated by: KAFKA-3423 can't start kafka connect with former configs in distributed mode
• links to: GitHub Pull Request #723