[KAFKA-6507] NPE in KafkaStatusBackingStore

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 0.11.0.1
• Component/s: connect
• Environment: We are using Kafka 0.10.0.1 with Kafka connect 0.11.0.1.
• Assignee: Unassigned
• Reporter: Itay Cohai
• Created: 31/Jan/18 13:43
• Updated: 06/Mar/18 07:51

Description

Found the following NPE in our kafka connect logs:

2018-01-30 13:15:34,391] ERROR Unexpected exception in Thread[KafkaBasedLog Work Thread - itay_test-connect-status,5,main] (org.apache.kafka.connect.util.KafkaBasedLog:334)

java.lang.NullPointerException

                at org.apache.kafka.connect.storage.KafkaStatusBackingStore.read(KafkaStatusBackingStore.java:441)

                at org.apache.kafka.connect.storage.KafkaStatusBackingStore$1.onCompletion(KafkaStatusBackingStore.java:148)

                at org.apache.kafka.connect.storage.KafkaStatusBackingStore$1.onCompletion(KafkaStatusBackingStore.java:145)

                at org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:258)

                at org.apache.kafka.connect.util.KafkaBasedLog.access$500(KafkaBasedLog.java:69)

                at org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:327)

 

If I look at the source, looks like the key comes up NULL from the status topic, strange.

void read(ConsumerRecord<String, byte[]> record) {
String key = record.key();

//This line --> if (key.startsWith(CONNECTOR_STATUS_PREFIX))

{ readConnectorStatus(key, record.value()); }

else if (key.startsWith(TASK_STATUS_PREFIX))

{ readTaskStatus(key, record.value()); }

else {
log.warn("Discarding record with invalid key {}", key);
}
}