tzulitai commented on a change in pull request #6537: [FLINK-10122]
KafkaConsumer should use partitionable state over union state if partition
discovery is not active
URL: https://github.com/apache/flink/pull/6537#discussion_r210808465
##########
File path:
flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
##########
@@ -783,30 +779,26 @@ public final void
initializeState(FunctionInitializationContext context) throws
OperatorStateStore stateStore = context.getOperatorStateStore();
- ListState<Tuple2<KafkaTopicPartition, Long>>
oldRoundRobinListState =
-
stateStore.getSerializableListState(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME);
+ final TypeInformation<Tuple2<KafkaTopicPartition, Long>>
offsetStateTypeInfo =
+ TypeInformation.of(new
TypeHint<Tuple2<KafkaTopicPartition, Long>>() {});
- this.unionOffsetStates = stateStore.getUnionListState(new
ListStateDescriptor<>(
- OFFSETS_STATE_NAME,
- TypeInformation.of(new
TypeHint<Tuple2<KafkaTopicPartition, Long>>() {})));
+ ListStateDescriptor<Tuple2<KafkaTopicPartition, Long>>
offsetStateDescriptor =
+ new ListStateDescriptor<>(OFFSETS_STATE_NAME,
offsetStateTypeInfo);
- if (context.isRestored() && !restoredFromOldState) {
- restoredState = new TreeMap<>(new
KafkaTopicPartition.Comparator());
+ this.offsetsState =
+ discoveryIntervalMillis != PARTITION_DISCOVERY_DISABLED
?
+
stateStore.getUnionListState(offsetStateDescriptor) :
stateStore.getListState(offsetStateDescriptor);
Review comment:
I think this is a "behaviour break".
Before, the user was free to switch between enabling / disabling partition
discovery on each restart for the job.
Now, with this change, we can't really do that, because changing the
distribution scheme of the same registered state isn't allowed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
