[FLINK-11987] Kafka producer occasionally throws NullpointerException

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.6.3, 1.6.4, 1.7.2
• Fix Version/s: 1.6.5, 1.7.3, 1.8.1, 1.9.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available

Environment

Flink 1.6.2 (Standalone Cluster)

Oracle JDK 1.8u151

Centos 7.4

• Assignee: Yu Li
• Reporter: LIU Xiao
• Created: 21/Mar/19 06:52
• Updated: 19/Feb/21 07:23
• Resolved: 02/Jun/19 05:49

Description

We are using Flink 1.6.2 in our production environment, and kafka producer occasionally throws NullpointerException.

We found in line 175 of flink/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java, NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR was created as a static variable.

Then in line 837,

context.getOperatorStateStore().getUnionListState(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);

was called, and that leads to line 734 of
flink/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java:

stateDescriptor.initializeSerializerUnlessSet(getExecutionConfig());

In function initializeSerializerUnlessSet(line 283 of flink/flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java):

if (serializer == null) {
  checkState(typeInfo != null, "no serializer and no type info");
  // instantiate the serializer
  serializer = typeInfo.createSerializer(executionConfig);
  // we can drop the type info now, no longer needed
  typeInfo  = null;
}

"serializer = typeInfo.createSerializer(executionConfig);" is the line which throws the exception.

We think that's because multiple subtasks of the same producer in a same TaskManager share a same NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR.

Issue Links

• is blocked by: FLINK-12688 Make serializer lazy initialization thread safe in StateDescriptor
• links to: GitHub Pull Request #8331