Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/4919#discussion_r147675623
--- Diff:
flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
---
@@ -993,14 +1000,162 @@ public String toString() {
* Context associated to this instance of the {@link
FlinkKafkaProducer011}. User for keeping track of the
* transactionalIds.
*/
- public static class KafkaTransactionContext {
- public final Set<String> transactionalIds;
+ static class KafkaTransactionContext {
+ final Set<String> transactionalIds;
- public KafkaTransactionContext(Set<String> transactionalIds) {
+ KafkaTransactionContext(Set<String> transactionalIds) {
--- End diff --
Here we should check for `null` for the `transactionalIds`. If this is
null, it will also break the `ContextStateSerializer.serialize()` method.
---
