You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "AHeise (via GitHub)" <gi...@apache.org> on 2025/09/30 13:50:52 UTC
[PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
AHeise opened a new pull request, #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192
KafkaEnumerator's state contains the TopicPartitions only but not the offsets, so it doesn't contain the full split state contrary to the design intent.
There are a couple of issues with that approach. It implicitly assumes that splits are fully assigned to readers before the first checkpoint. Else the enumerator will invoke the offset initializer again on recovery from such a checkpoint leading to inconsistencies (LATEST may be initialized during the first attempt for some partitions and initialized during second attempt for others).
Through addSplitBack callback, you may also get these scenarios later for BATCH which actually leads to duplicate rows (in case of EARLIEST or SPECIFIC-OFFSETS) or data loss (in case of LATEST). Finally, it's not possible to safely use KafkaSource as part of a HybridSource because the offset initializer cannot even be recreated on recovery.
All cases are solved by also retaining the offset in the enumerator state. To that end, this commit merges the async discovery phases to immediately initialize the splits from the partitions. Any subsequent checkpoint will contain the proper start offset.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#issuecomment-3377452417
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2408597745
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -316,14 +329,20 @@ public void testDiscoverPartitionsPeriodically() throws Throwable {
}
}
- @Test
- public void testAddSplitsBack() throws Throwable {
+ @ParameterizedTest
+ @EnumSource(StandardOffsetsInitializer.class)
+ public void testAddSplitsBack(StandardOffsetsInitializer offsetsInitializer) throws Throwable {
Review Comment:
Good catch. I expanded the test to cover snapshotting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2392448704
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializerTest.java:
##########
@@ -56,73 +58,108 @@ public void testEnumStateSerde() throws IOException {
final KafkaSourceEnumState restoredState =
serializer.deserialize(serializer.getVersion(), bytes);
- assertThat(restoredState.assignedPartitions()).isEqualTo(state.assignedPartitions());
- assertThat(restoredState.unassignedInitialPartitions())
- .isEqualTo(state.unassignedInitialPartitions());
+ assertThat(restoredState.assignedSplits())
+ .containsExactlyInAnyOrderElementsOf(state.assignedSplits());
+ assertThat(restoredState.unassignedSplits())
+ .containsExactlyInAnyOrderElementsOf(state.unassignedSplits());
assertThat(restoredState.initialDiscoveryFinished()).isTrue();
}
@Test
public void testBackwardCompatibility() throws IOException {
- final Set<TopicPartition> topicPartitions = constructTopicPartitions(0);
- final Map<Integer, Set<KafkaPartitionSplit>> splitAssignments =
- toSplitAssignments(topicPartitions);
+ final Set<KafkaPartitionSplit> splits = constructTopicSplits(0);
+ final Map<Integer, Collection<KafkaPartitionSplit>> splitAssignments =
+ toSplitAssignments(splits);
+ final List<SplitAndAssignmentStatus> splitAndAssignmentStatuses =
+ splits.stream()
+ .map(
+ split ->
+ new SplitAndAssignmentStatus(
+ split, getAssignmentStatus(split)))
+ .collect(Collectors.toList());
// Create bytes in the way of KafkaEnumStateSerializer version 0 doing serialization
final byte[] bytesV0 =
SerdeUtils.serializeSplitAssignments(
splitAssignments, new KafkaPartitionSplitSerializer());
// Create bytes in the way of KafkaEnumStateSerializer version 1 doing serialization
- final byte[] bytesV1 =
- KafkaSourceEnumStateSerializer.serializeTopicPartitions(topicPartitions);
+ final byte[] bytesV1 = KafkaSourceEnumStateSerializer.serializeV1(splits);
+ final byte[] bytesV2 =
+ KafkaSourceEnumStateSerializer.serializeV2(splitAndAssignmentStatuses, false);
// Deserialize above bytes with KafkaEnumStateSerializer version 2 to check backward
// compatibility
final KafkaSourceEnumState kafkaSourceEnumStateV0 =
new KafkaSourceEnumStateSerializer().deserialize(0, bytesV0);
final KafkaSourceEnumState kafkaSourceEnumStateV1 =
new KafkaSourceEnumStateSerializer().deserialize(1, bytesV1);
+ final KafkaSourceEnumState kafkaSourceEnumStateV2 =
+ new KafkaSourceEnumStateSerializer().deserialize(2, bytesV2);
- assertThat(kafkaSourceEnumStateV0.assignedPartitions()).isEqualTo(topicPartitions);
- assertThat(kafkaSourceEnumStateV0.unassignedInitialPartitions()).isEmpty();
+ assertThat(kafkaSourceEnumStateV0.assignedSplits())
+ .containsExactlyInAnyOrderElementsOf(splits);
+ assertThat(kafkaSourceEnumStateV0.unassignedSplits()).isEmpty();
assertThat(kafkaSourceEnumStateV0.initialDiscoveryFinished()).isTrue();
- assertThat(kafkaSourceEnumStateV1.assignedPartitions()).isEqualTo(topicPartitions);
- assertThat(kafkaSourceEnumStateV1.unassignedInitialPartitions()).isEmpty();
+ assertThat(kafkaSourceEnumStateV1.assignedSplits())
+ .containsExactlyInAnyOrderElementsOf(splits);
+ assertThat(kafkaSourceEnumStateV1.unassignedSplits()).isEmpty();
assertThat(kafkaSourceEnumStateV1.initialDiscoveryFinished()).isTrue();
+
+ final Map<AssignmentStatus, Set<KafkaPartitionSplit>> splitsByStatus =
+ splitAndAssignmentStatuses.stream()
+ .collect(
+ Collectors.groupingBy(
+ SplitAndAssignmentStatus::assignmentStatus,
+ Collectors.mapping(
+ SplitAndAssignmentStatus::split,
+ Collectors.toSet())));
+ assertThat(kafkaSourceEnumStateV2.assignedSplits())
+ .containsExactlyInAnyOrderElementsOf(splitsByStatus.get(AssignmentStatus.ASSIGNED));
+ assertThat(kafkaSourceEnumStateV2.unassignedSplits())
+ .containsExactlyInAnyOrderElementsOf(
+ splitsByStatus.get(AssignmentStatus.UNASSIGNED));
+ assertThat(kafkaSourceEnumStateV2.initialDiscoveryFinished()).isFalse();
+ }
+
+ private static AssignmentStatus getAssignmentStatus(KafkaPartitionSplit split) {
+ return AssignmentStatus.values()[
+ Math.abs(split.hashCode()) % AssignmentStatus.values().length];
}
- private Set<TopicPartition> constructTopicPartitions(int startPartition) {
+ private Set<KafkaPartitionSplit> constructTopicSplits(int startPartition) {
// Create topic partitions for readers.
// Reader i will be assigned with NUM_PARTITIONS_PER_TOPIC splits, with topic name
// "topic-{i}" and
// NUM_PARTITIONS_PER_TOPIC partitions. The starting partition number is startPartition
// Totally NUM_READERS * NUM_PARTITIONS_PER_TOPIC partitions will be created.
- Set<TopicPartition> topicPartitions = new HashSet<>();
+ Set<KafkaPartitionSplit> topicPartitions = new HashSet<>();
for (int readerId = 0; readerId < NUM_READERS; readerId++) {
for (int partition = startPartition;
partition < startPartition + NUM_PARTITIONS_PER_TOPIC;
partition++) {
- topicPartitions.add(new TopicPartition(TOPIC_PREFIX + readerId, partition));
+ topicPartitions.add(
+ new KafkaPartitionSplit(
+ new TopicPartition(TOPIC_PREFIX + readerId, partition),
+ STARTING_OFFSET));
Review Comment:
Thanks for the PR. This is a very good improvement for the connector.
I noticed that the current test creates splits using the constant `KafkaPartitionSplit.EARLIEST_OFFSET`, would it make sense to add a test case that uses a real-world offset (e.g., `123`)?
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -316,14 +329,20 @@ public void testDiscoverPartitionsPeriodically() throws Throwable {
}
}
- @Test
- public void testAddSplitsBack() throws Throwable {
+ @ParameterizedTest
+ @EnumSource(StandardOffsetsInitializer.class)
+ public void testAddSplitsBack(StandardOffsetsInitializer offsetsInitializer) throws Throwable {
Review Comment:
Is my understanding correct that the test verifies that the offset is correctly recalculated on recovery, but doesn't verify that the original offset(before the failure) was preserved and restored?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2409593282
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -229,38 +239,16 @@ public void close() {
*
* @return Set of subscribed {@link TopicPartition}s
*/
- private Set<TopicPartition> getSubscribedTopicPartitions() {
- return subscriber.getSubscribedTopicPartitions(adminClient);
- }
-
- /**
- * Check if there's any partition changes within subscribed topic partitions fetched by worker
- * thread, and invoke {@link KafkaSourceEnumerator#initializePartitionSplits(PartitionChange)}
- * in worker thread to initialize splits for new partitions.
- *
- * <p>NOTE: This method should only be invoked in the coordinator executor thread.
- *
- * @param fetchedPartitions Map from topic name to its description
- * @param t Exception in worker thread
- */
- private void checkPartitionChanges(Set<TopicPartition> fetchedPartitions, Throwable t) {
Review Comment:
in hindsight, yes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise merged PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2408539975
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializerTest.java:
##########
@@ -56,73 +58,108 @@ public void testEnumStateSerde() throws IOException {
final KafkaSourceEnumState restoredState =
serializer.deserialize(serializer.getVersion(), bytes);
- assertThat(restoredState.assignedPartitions()).isEqualTo(state.assignedPartitions());
- assertThat(restoredState.unassignedInitialPartitions())
- .isEqualTo(state.unassignedInitialPartitions());
+ assertThat(restoredState.assignedSplits())
+ .containsExactlyInAnyOrderElementsOf(state.assignedSplits());
+ assertThat(restoredState.unassignedSplits())
+ .containsExactlyInAnyOrderElementsOf(state.unassignedSplits());
assertThat(restoredState.initialDiscoveryFinished()).isTrue();
}
@Test
public void testBackwardCompatibility() throws IOException {
- final Set<TopicPartition> topicPartitions = constructTopicPartitions(0);
- final Map<Integer, Set<KafkaPartitionSplit>> splitAssignments =
- toSplitAssignments(topicPartitions);
+ final Set<KafkaPartitionSplit> splits = constructTopicSplits(0);
+ final Map<Integer, Collection<KafkaPartitionSplit>> splitAssignments =
+ toSplitAssignments(splits);
+ final List<SplitAndAssignmentStatus> splitAndAssignmentStatuses =
+ splits.stream()
+ .map(
+ split ->
+ new SplitAndAssignmentStatus(
+ split, getAssignmentStatus(split)))
+ .collect(Collectors.toList());
// Create bytes in the way of KafkaEnumStateSerializer version 0 doing serialization
final byte[] bytesV0 =
SerdeUtils.serializeSplitAssignments(
splitAssignments, new KafkaPartitionSplitSerializer());
// Create bytes in the way of KafkaEnumStateSerializer version 1 doing serialization
- final byte[] bytesV1 =
- KafkaSourceEnumStateSerializer.serializeTopicPartitions(topicPartitions);
+ final byte[] bytesV1 = KafkaSourceEnumStateSerializer.serializeV1(splits);
+ final byte[] bytesV2 =
+ KafkaSourceEnumStateSerializer.serializeV2(splitAndAssignmentStatuses, false);
// Deserialize above bytes with KafkaEnumStateSerializer version 2 to check backward
// compatibility
final KafkaSourceEnumState kafkaSourceEnumStateV0 =
new KafkaSourceEnumStateSerializer().deserialize(0, bytesV0);
final KafkaSourceEnumState kafkaSourceEnumStateV1 =
new KafkaSourceEnumStateSerializer().deserialize(1, bytesV1);
+ final KafkaSourceEnumState kafkaSourceEnumStateV2 =
+ new KafkaSourceEnumStateSerializer().deserialize(2, bytesV2);
- assertThat(kafkaSourceEnumStateV0.assignedPartitions()).isEqualTo(topicPartitions);
- assertThat(kafkaSourceEnumStateV0.unassignedInitialPartitions()).isEmpty();
+ assertThat(kafkaSourceEnumStateV0.assignedSplits())
+ .containsExactlyInAnyOrderElementsOf(splits);
+ assertThat(kafkaSourceEnumStateV0.unassignedSplits()).isEmpty();
assertThat(kafkaSourceEnumStateV0.initialDiscoveryFinished()).isTrue();
- assertThat(kafkaSourceEnumStateV1.assignedPartitions()).isEqualTo(topicPartitions);
- assertThat(kafkaSourceEnumStateV1.unassignedInitialPartitions()).isEmpty();
+ assertThat(kafkaSourceEnumStateV1.assignedSplits())
+ .containsExactlyInAnyOrderElementsOf(splits);
+ assertThat(kafkaSourceEnumStateV1.unassignedSplits()).isEmpty();
assertThat(kafkaSourceEnumStateV1.initialDiscoveryFinished()).isTrue();
+
+ final Map<AssignmentStatus, Set<KafkaPartitionSplit>> splitsByStatus =
+ splitAndAssignmentStatuses.stream()
+ .collect(
+ Collectors.groupingBy(
+ SplitAndAssignmentStatus::assignmentStatus,
+ Collectors.mapping(
+ SplitAndAssignmentStatus::split,
+ Collectors.toSet())));
+ assertThat(kafkaSourceEnumStateV2.assignedSplits())
+ .containsExactlyInAnyOrderElementsOf(splitsByStatus.get(AssignmentStatus.ASSIGNED));
+ assertThat(kafkaSourceEnumStateV2.unassignedSplits())
+ .containsExactlyInAnyOrderElementsOf(
+ splitsByStatus.get(AssignmentStatus.UNASSIGNED));
+ assertThat(kafkaSourceEnumStateV2.initialDiscoveryFinished()).isFalse();
+ }
+
+ private static AssignmentStatus getAssignmentStatus(KafkaPartitionSplit split) {
+ return AssignmentStatus.values()[
+ Math.abs(split.hashCode()) % AssignmentStatus.values().length];
}
- private Set<TopicPartition> constructTopicPartitions(int startPartition) {
+ private Set<KafkaPartitionSplit> constructTopicSplits(int startPartition) {
// Create topic partitions for readers.
// Reader i will be assigned with NUM_PARTITIONS_PER_TOPIC splits, with topic name
// "topic-{i}" and
// NUM_PARTITIONS_PER_TOPIC partitions. The starting partition number is startPartition
// Totally NUM_READERS * NUM_PARTITIONS_PER_TOPIC partitions will be created.
- Set<TopicPartition> topicPartitions = new HashSet<>();
+ Set<KafkaPartitionSplit> topicPartitions = new HashSet<>();
for (int readerId = 0; readerId < NUM_READERS; readerId++) {
for (int partition = startPartition;
partition < startPartition + NUM_PARTITIONS_PER_TOPIC;
partition++) {
- topicPartitions.add(new TopicPartition(TOPIC_PREFIX + readerId, partition));
+ topicPartitions.add(
+ new KafkaPartitionSplit(
+ new TopicPartition(TOPIC_PREFIX + readerId, partition),
+ STARTING_OFFSET));
Review Comment:
I had to change the logic a bit and introduced a new special value MIGRATED against which all unit tests now go. However, I also added a test with specific offsets to `KafkaSourceEnumeratorTest`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2409595432
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -662,63 +688,84 @@ private void verifyLastReadersAssignments(
Collection<Integer> readers,
Set<String> topics,
int expectedAssignmentSeqSize) {
+ verifyLastReadersAssignments(
+ context, readers, topics, expectedAssignmentSeqSize, OffsetsInitializer.earliest());
+ }
+
+ private void verifyLastReadersAssignments(
+ MockSplitEnumeratorContext<KafkaPartitionSplit> context,
+ Collection<Integer> readers,
+ Set<String> topics,
+ int expectedAssignmentSeqSize,
+ OffsetsInitializer offsetsInitializer) {
verifyAssignments(
- getExpectedAssignments(new HashSet<>(readers), topics),
+ getExpectedAssignments(new HashSet<>(readers), topics, offsetsInitializer),
context.getSplitsAssignmentSequence()
.get(expectedAssignmentSeqSize - 1)
.assignment());
}
private void verifyAssignments(
- Map<Integer, Set<TopicPartition>> expectedAssignments,
+ Map<Integer, Collection<KafkaPartitionSplit>> expectedAssignments,
Map<Integer, List<KafkaPartitionSplit>> actualAssignments) {
- actualAssignments.forEach(
- (reader, splits) -> {
- Set<TopicPartition> expectedAssignmentsForReader =
- expectedAssignments.get(reader);
- assertThat(expectedAssignmentsForReader).isNotNull();
- assertThat(splits.size()).isEqualTo(expectedAssignmentsForReader.size());
- for (KafkaPartitionSplit split : splits) {
- assertThat(expectedAssignmentsForReader)
- .contains(split.getTopicPartition());
+ assertThat(actualAssignments).containsOnlyKeys(expectedAssignments.keySet());
+ SoftAssertions.assertSoftly(
Review Comment:
Here I use soft assertions for assertion on the nested structure of the same data structure. I haven't found a good way to have all assertions on this one data structure in one line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2408038369
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -132,22 +197,43 @@ private static KafkaSourceEnumState deserializeAssignedTopicPartitions(
}
}
- private static KafkaSourceEnumState deserializeTopicPartitionAndAssignmentStatus(
- byte[] serialized) throws IOException {
+ @VisibleForTesting
+ static byte[] serializeV2(
+ Collection<SplitAndAssignmentStatus> splits, boolean initialDiscoveryFinished)
+ throws IOException {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+ out.writeInt(splits.size());
+ for (SplitAndAssignmentStatus splitAndAssignmentStatus : splits) {
+ final TopicPartition topicPartition =
+ splitAndAssignmentStatus.split().getTopicPartition();
+ out.writeUTF(topicPartition.topic());
+ out.writeInt(topicPartition.partition());
+ out.writeInt(splitAndAssignmentStatus.assignmentStatus().getStatusCode());
+ }
+ out.writeBoolean(initialDiscoveryFinished);
+ out.flush();
+ return baos.toByteArray();
+ }
+ }
+
+ private static KafkaSourceEnumState deserializeVersion2(byte[] serialized) throws IOException {
try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
DataInputStream in = new DataInputStream(bais)) {
final int numPartitions = in.readInt();
- Set<TopicPartitionAndAssignmentStatus> partitions = new HashSet<>(numPartitions);
+ Set<SplitAndAssignmentStatus> partitions = new HashSet<>(numPartitions);
for (int i = 0; i < numPartitions; i++) {
final String topic = in.readUTF();
final int partition = in.readInt();
final int statusCode = in.readInt();
partitions.add(
- new TopicPartitionAndAssignmentStatus(
- new TopicPartition(topic, partition),
+ new SplitAndAssignmentStatus(
+ new KafkaPartitionSplit(
+ new TopicPartition(topic, partition),
+ DEFAULT_STARTING_OFFSET),
Review Comment:
Yes, added a new MIGRATED offset to indicate that this needs to be initialized on recovery.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2397330827
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -132,22 +197,43 @@ private static KafkaSourceEnumState deserializeAssignedTopicPartitions(
}
}
- private static KafkaSourceEnumState deserializeTopicPartitionAndAssignmentStatus(
- byte[] serialized) throws IOException {
+ @VisibleForTesting
+ static byte[] serializeV2(
+ Collection<SplitAndAssignmentStatus> splits, boolean initialDiscoveryFinished)
+ throws IOException {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+ out.writeInt(splits.size());
+ for (SplitAndAssignmentStatus splitAndAssignmentStatus : splits) {
+ final TopicPartition topicPartition =
+ splitAndAssignmentStatus.split().getTopicPartition();
+ out.writeUTF(topicPartition.topic());
+ out.writeInt(topicPartition.partition());
+ out.writeInt(splitAndAssignmentStatus.assignmentStatus().getStatusCode());
+ }
+ out.writeBoolean(initialDiscoveryFinished);
+ out.flush();
+ return baos.toByteArray();
+ }
+ }
+
+ private static KafkaSourceEnumState deserializeVersion2(byte[] serialized) throws IOException {
try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
DataInputStream in = new DataInputStream(bais)) {
final int numPartitions = in.readInt();
- Set<TopicPartitionAndAssignmentStatus> partitions = new HashSet<>(numPartitions);
+ Set<SplitAndAssignmentStatus> partitions = new HashSet<>(numPartitions);
for (int i = 0; i < numPartitions; i++) {
final String topic = in.readUTF();
final int partition = in.readInt();
final int statusCode = in.readInt();
partitions.add(
- new TopicPartitionAndAssignmentStatus(
- new TopicPartition(topic, partition),
+ new SplitAndAssignmentStatus(
+ new KafkaPartitionSplit(
+ new TopicPartition(topic, partition),
+ DEFAULT_STARTING_OFFSET),
Review Comment:
Isn't this a behavioral change? Previously the unassigned split would get the starting offset configured by the user on reassignment.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -662,63 +688,84 @@ private void verifyLastReadersAssignments(
Collection<Integer> readers,
Set<String> topics,
int expectedAssignmentSeqSize) {
+ verifyLastReadersAssignments(
+ context, readers, topics, expectedAssignmentSeqSize, OffsetsInitializer.earliest());
+ }
+
+ private void verifyLastReadersAssignments(
+ MockSplitEnumeratorContext<KafkaPartitionSplit> context,
+ Collection<Integer> readers,
+ Set<String> topics,
+ int expectedAssignmentSeqSize,
+ OffsetsInitializer offsetsInitializer) {
verifyAssignments(
- getExpectedAssignments(new HashSet<>(readers), topics),
+ getExpectedAssignments(new HashSet<>(readers), topics, offsetsInitializer),
context.getSplitsAssignmentSequence()
.get(expectedAssignmentSeqSize - 1)
.assignment());
}
private void verifyAssignments(
- Map<Integer, Set<TopicPartition>> expectedAssignments,
+ Map<Integer, Collection<KafkaPartitionSplit>> expectedAssignments,
Map<Integer, List<KafkaPartitionSplit>> actualAssignments) {
- actualAssignments.forEach(
- (reader, splits) -> {
- Set<TopicPartition> expectedAssignmentsForReader =
- expectedAssignments.get(reader);
- assertThat(expectedAssignmentsForReader).isNotNull();
- assertThat(splits.size()).isEqualTo(expectedAssignmentsForReader.size());
- for (KafkaPartitionSplit split : splits) {
- assertThat(expectedAssignmentsForReader)
- .contains(split.getTopicPartition());
+ assertThat(actualAssignments).containsOnlyKeys(expectedAssignments.keySet());
+ SoftAssertions.assertSoftly(
Review Comment:
I am not a big fan of soft assertions but this pattern with the lambda I think is "okay". It's just a dangerous precedent when folks start using the the version that requires explicitly calling `assertAll()` and we silently disable all exceptions.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -96,7 +96,7 @@ public class KafkaSourceEnumerator
// initializing partition discovery has finished.
private boolean noMoreNewPartitionSplits = false;
// this flag will be marked as true if initial partitions are discovered after enumerator starts
- private boolean initialDiscoveryFinished;
+ private volatile boolean initialDiscoveryFinished;
Review Comment:
Can you explain why volatile is needed here? Afaik we don't access it from a different thread than before.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -229,38 +239,16 @@ public void close() {
*
* @return Set of subscribed {@link TopicPartition}s
*/
- private Set<TopicPartition> getSubscribedTopicPartitions() {
- return subscriber.getSubscribedTopicPartitions(adminClient);
- }
-
- /**
- * Check if there's any partition changes within subscribed topic partitions fetched by worker
- * thread, and invoke {@link KafkaSourceEnumerator#initializePartitionSplits(PartitionChange)}
- * in worker thread to initialize splits for new partitions.
- *
- * <p>NOTE: This method should only be invoked in the coordinator executor thread.
- *
- * @param fetchedPartitions Map from topic name to its description
- * @param t Exception in worker thread
- */
- private void checkPartitionChanges(Set<TopicPartition> fetchedPartitions, Throwable t) {
Review Comment:
Nit: It would have been good to have a small refactoring commit to simplify the method calls it makes the reviewer easier :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2409626057
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -131,19 +134,84 @@ public KafkaSourceEnumerator(
this.context = context;
this.boundedness = boundedness;
- this.assignedPartitions = new HashSet<>(kafkaSourceEnumState.assignedPartitions());
+ Map<AssignmentStatus, List<KafkaPartitionSplit>> splits =
+ initializeMigratedSplits(kafkaSourceEnumState.splits());
+ this.assignedSplits = indexByPartition(splits.get(AssignmentStatus.ASSIGNED));
+ this.unassignedSplits = indexByPartition(splits.get(AssignmentStatus.UNASSIGNED));
this.pendingPartitionSplitAssignment = new HashMap<>();
this.partitionDiscoveryIntervalMs =
KafkaSourceOptions.getOption(
properties,
KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS,
Long::parseLong);
this.consumerGroupId = properties.getProperty(ConsumerConfig.GROUP_ID_CONFIG);
- this.unassignedInitialPartitions =
- new HashSet<>(kafkaSourceEnumState.unassignedInitialPartitions());
this.initialDiscoveryFinished = kafkaSourceEnumState.initialDiscoveryFinished();
}
+ /**
+ * Initialize migrated splits to splits with concrete starting offsets. This method ensures that
+ * the costly offset resolution is performed only when there are splits that have been
+ * checkpointed with previous enumerator versions.
+ *
+ * <p>Note that this method is deliberately performed in the main thread to avoid a checkpoint
+ * of the splits without starting offset.
+ */
+ private Map<AssignmentStatus, List<KafkaPartitionSplit>> initializeMigratedSplits(
+ Set<SplitAndAssignmentStatus> splits) {
+ final Set<TopicPartition> migratedPartitions =
+ splits.stream()
+ .filter(
+ splitStatus ->
+ splitStatus.split().getStartingOffset()
+ == KafkaPartitionSplit.MIGRATED)
+ .map(splitStatus -> splitStatus.split().getTopicPartition())
+ .collect(Collectors.toSet());
+
+ if (migratedPartitions.isEmpty()) {
+ return splitByAssignmentStatus(splits.stream());
+ }
+
+ final Map<TopicPartition, Long> startOffsets =
+ startingOffsetInitializer.getPartitionOffsets(
+ migratedPartitions, getOffsetsRetriever());
+ return splitByAssignmentStatus(
+ splits.stream()
+ .map(splitStatus -> resolveMigratedSplit(splitStatus, startOffsets)));
Review Comment:
Nit: The flow of extracting the migratedPartitions is overly complex because we extract the migrated partitions twice in line 179 and line 161.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -316,39 +341,67 @@ public void testDiscoverPartitionsPeriodically() throws Throwable {
}
}
- @Test
- public void testAddSplitsBack() throws Throwable {
+ @ParameterizedTest
+ @EnumSource(StandardOffsetsInitializer.class)
+ public void testAddSplitsBack(StandardOffsetsInitializer offsetsInitializer) throws Throwable {
Review Comment:
Can you also add a test to cover the newly added migration story?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2409592648
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -96,7 +96,7 @@ public class KafkaSourceEnumerator
// initializing partition discovery has finished.
private boolean noMoreNewPartitionSplits = false;
// this flag will be marked as true if initial partitions are discovered after enumerator starts
- private boolean initialDiscoveryFinished;
+ private volatile boolean initialDiscoveryFinished;
Review Comment:
I added a comment explaining the access pattern. The short answer is that I haven't found a way to inject the current state of variable into the worker thread. I could schedule the first discovery differently with `context.callAsync(() -> discover(true));` and then schedule periodic discovery differently but that's a bigger change.
Since it's not on the hot path whatsoever, I'd leave it on a better documented volatile.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38453] Add full splits to KafkaSourceEnumState [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #192:
URL: https://github.com/apache/flink-connector-kafka/pull/192#discussion_r2409711198
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -131,19 +134,84 @@ public KafkaSourceEnumerator(
this.context = context;
this.boundedness = boundedness;
- this.assignedPartitions = new HashSet<>(kafkaSourceEnumState.assignedPartitions());
+ Map<AssignmentStatus, List<KafkaPartitionSplit>> splits =
+ initializeMigratedSplits(kafkaSourceEnumState.splits());
+ this.assignedSplits = indexByPartition(splits.get(AssignmentStatus.ASSIGNED));
+ this.unassignedSplits = indexByPartition(splits.get(AssignmentStatus.UNASSIGNED));
this.pendingPartitionSplitAssignment = new HashMap<>();
this.partitionDiscoveryIntervalMs =
KafkaSourceOptions.getOption(
properties,
KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS,
Long::parseLong);
this.consumerGroupId = properties.getProperty(ConsumerConfig.GROUP_ID_CONFIG);
- this.unassignedInitialPartitions =
- new HashSet<>(kafkaSourceEnumState.unassignedInitialPartitions());
this.initialDiscoveryFinished = kafkaSourceEnumState.initialDiscoveryFinished();
}
+ /**
+ * Initialize migrated splits to splits with concrete starting offsets. This method ensures that
+ * the costly offset resolution is performed only when there are splits that have been
+ * checkpointed with previous enumerator versions.
+ *
+ * <p>Note that this method is deliberately performed in the main thread to avoid a checkpoint
+ * of the splits without starting offset.
+ */
+ private Map<AssignmentStatus, List<KafkaPartitionSplit>> initializeMigratedSplits(
+ Set<SplitAndAssignmentStatus> splits) {
+ final Set<TopicPartition> migratedPartitions =
+ splits.stream()
+ .filter(
+ splitStatus ->
+ splitStatus.split().getStartingOffset()
+ == KafkaPartitionSplit.MIGRATED)
+ .map(splitStatus -> splitStatus.split().getTopicPartition())
+ .collect(Collectors.toSet());
+
+ if (migratedPartitions.isEmpty()) {
+ return splitByAssignmentStatus(splits.stream());
+ }
+
+ final Map<TopicPartition, Long> startOffsets =
+ startingOffsetInitializer.getPartitionOffsets(
+ migratedPartitions, getOffsetsRetriever());
+ return splitByAssignmentStatus(
+ splits.stream()
+ .map(splitStatus -> resolveMigratedSplit(splitStatus, startOffsets)));
Review Comment:
It's unfortunately necessary by design:
* 161 extracts the partitions which are used to jointly look up the partition offsets
* This is expensive as it uses admin client to contact Kafka cluster
* The design of offset initializer is to jointly look up all partitions to have 1 request to Kafka brokers only
* Now that we received all offsets, 179 is applying them to the split. It could be a simple map lookup but I decided to add some assertion, so it went into a different method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org