From 944a63cf4a6c8ae2b38bbb12b2ba1103407afea7 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Mon, 30 Oct 2017 13:46:37 +0100 Subject: [PATCH] [hotfix] Make fields transient in TwoPhaseCommitSinkFunction --- .../streaming/connectors/kafka/FlinkKafkaProducer011.java | 2 +- .../api/functions/sink/TwoPhaseCommitSinkFunction.java | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java index ed3a31daf0fe9..593e002de5494 100644 --- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java +++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java @@ -179,7 +179,7 @@ public enum Semantic { /** * Hint for picking next transactional id. */ - private NextTransactionalIdHint nextTransactionalIdHint; + private transient NextTransactionalIdHint nextTransactionalIdHint; /** * User defined properties for the Producer. diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java index 2dfa292b76251..8c117535d3eb0 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java @@ -65,10 +65,10 @@ public abstract class TwoPhaseCommitSinkFunction protected final LinkedHashMap pendingCommitTransactions = new LinkedHashMap<>(); @Nullable - protected TXN currentTransaction; - protected Optional userContext; + protected transient TXN currentTransaction; + protected transient Optional userContext; - protected ListState> state; + protected transient ListState> state; /** * Use default {@link ListStateDescriptor} for internal state serialization. Helpful utilities for using this