You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Chethan U (JIRA)" <ji...@apache.org> on 2019/05/13 15:25:00 UTC
[jira] [Comment Edited] (FLINK-11987) Kafka producer occasionally
throws NullpointerException
[ https://issues.apache.org/jira/browse/FLINK-11987?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16838541#comment-16838541 ]
Chethan U edited comment on FLINK-11987 at 5/13/19 3:24 PM:
------------------------------------------------------------
Yes, I also faced this issue. Out of nowhere some NullpointerException.
Error:
java.lang.NullPointerException
at java.io.StringReader.<init>(StringReader.java:50)
at com.google.gson.JsonParser.parse(JsonParser.java:45)
at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:718)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:696)
at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:718)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:696)
at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111)
at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordWithTimestamp(AbstractFetcher.java:398)
at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.emitRecord(KafkaFetcher.java:185)
at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:150)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:711)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:93)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:57)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:97)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
at java.lang.Thread.run(Thread.java:748)
was (Author: chethanuk):
Yes, I also faced this issue. Out of nowhere some NullpointerException.
> Kafka producer occasionally throws NullpointerException
> -------------------------------------------------------
>
> Key: FLINK-11987
> URL: https://issues.apache.org/jira/browse/FLINK-11987
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 1.6.3, 1.6.4, 1.7.2
> Environment: Flink 1.6.2 (Standalone Cluster)
> Oracle JDK 1.8u151
> Centos 7.4
> Reporter: LIU Xiao
> Priority: Blocker
> Labels: pull-request-available
> Fix For: 1.7.3, 1.6.5, 1.8.1
>
> Time Spent: 10m
> Remaining Estimate: 0h
>
> 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,
> {code:java}
> context.getOperatorStateStore().getUnionListState(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
> {code}
> was called, and that leads to line 734 of
> flink/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java:
> {code:java}
> stateDescriptor.initializeSerializerUnlessSet(getExecutionConfig());
> {code}
> In function initializeSerializerUnlessSet(line 283 of flink/flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java):
> {code: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.
> {code}
> We think that's because multiple subtasks of the same producer in a same TaskManager share a same NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)