You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2023/05/08 04:22:40 UTC
[GitHub] [flink-connector-kafka] loserwang1024 opened a new pull request, #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
loserwang1024 opened a new pull request, #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28
### What is the purpose of the change
As described in [[FLIP-288](https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source)](https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source), the strategy used for new partitions is the same as the initial offset strategy, which is not reasonable.
According to the semantics, if the startup strategy is latest, the consumed data should include all data from the moment of startup, which also includes all messages from new created partitions. However, the latest strategy currently maybe used for new partitions, leading to the loss of some data (thinking a new partition is created and might be discovered by Kafka source several minutes later, and the message produced into the partition within the gap might be dropped if we use for example "latest" as the initial offset strategy).if the data from all new partitions is not read, it does not meet the user's expectations.
Other ploblems see final Section: `User specifies OffsetsInitializer for new partition` .
Therefore, it’s better to provide an **EARLIEST** strategy for later discovered partitions.
### Brief change log
1. Expand `KafkaSourceEnumState` with `TopicPartitionWithAssignStatus` to distinguish between initial partitions and newly discovered partitions. `TopicPartitionWithAssignStatus` is also better for future expansion, as new statuses can be added without changing the state results.
2. Add a `newDiscoveryOffsetsInitializer`(EARLIEST) to get offsets for newly discovered partitions.
3. Modify `kafkaSourceEnumStateSerializer` to handle the expanded `KafkaSourceEnumState`.
### Verifying this change
1. Test the backward compatibility of state when deserializing in `KafkaSourceEnumStateSerializerTest`.
2. Expand `KafkaEnumeratorTest#testSnapshotState` method to test snapshot state in more scenarios:
1. Before first discovery, so the state should be empty
2. First partition discovery after start, but no assignments to readers
3. Assign partials partitions to readers
4. Assign all partitions to readers
3. Expand `KafkaEnumeratorTest#testDiscoverPartitionsPeriodically` method to test whether new partitions use EARLIEST offset while initial partitions use specified offset strategy.
--
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
[GitHub] [flink-connector-kafka] loserwang1024 commented on pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#issuecomment-1567676029
Now that @RamanVerma is busy, could anyone else help me? CC, @PatrickRen
--
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
[GitHub] [flink-connector-kafka] RamanVerma commented on a diff in pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "RamanVerma (via GitHub)" <gi...@apache.org>.
RamanVerma commented on code in PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#discussion_r1190292883
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumState.java:
##########
@@ -22,18 +22,69 @@
import org.apache.kafka.common.TopicPartition;
+import java.util.HashSet;
import java.util.Set;
+import java.util.stream.Collectors;
/** The state of Kafka source enumerator. */
@Internal
public class KafkaSourceEnumState {
- private final Set<TopicPartition> assignedPartitions;
+ /** Partitions with status: ASSIGNED or UNASSIGNED_INITIAL. */
+ private final Set<TopicPartitionWithAssignStatus> partitions;
+ /**
+ * this flag will be marked as true if inital partitions are discovered after enumerator starts.
+ */
+ private final boolean initialDiscoveryFinished;
- KafkaSourceEnumState(Set<TopicPartition> assignedPartitions) {
- this.assignedPartitions = assignedPartitions;
+ KafkaSourceEnumState(
+ Set<TopicPartition> assignPartitions,
Review Comment:
please change the parameter names to `assignedPartitions` and `unassignedInitialPartitions`
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/TopicPartitionWithAssignStatus.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.source.enumerator;
+
+import org.apache.flink.annotation.Internal;
+
+import org.apache.kafka.common.TopicPartition;
+
+/** Kafka partition with assign status. */
+@Internal
+public class TopicPartitionWithAssignStatus {
+ private final TopicPartition topicPartition;
+ private final long assignStatus;
Review Comment:
`assignmentStatus` would convey the meaning better than `assignStatus`
Also, I would prefer `TopicPartitionAndAssignmentStatus` over `TopicPartitionWithAssignStatus`
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumState.java:
##########
@@ -22,18 +22,69 @@
import org.apache.kafka.common.TopicPartition;
+import java.util.HashSet;
import java.util.Set;
+import java.util.stream.Collectors;
/** The state of Kafka source enumerator. */
@Internal
public class KafkaSourceEnumState {
- private final Set<TopicPartition> assignedPartitions;
+ /** Partitions with status: ASSIGNED or UNASSIGNED_INITIAL. */
+ private final Set<TopicPartitionWithAssignStatus> partitions;
+ /**
+ * this flag will be marked as true if inital partitions are discovered after enumerator starts.
+ */
+ private final boolean initialDiscoveryFinished;
- KafkaSourceEnumState(Set<TopicPartition> assignedPartitions) {
- this.assignedPartitions = assignedPartitions;
+ KafkaSourceEnumState(
+ Set<TopicPartition> assignPartitions,
+ Set<TopicPartition> unAssignInitialPartitions,
+ boolean initialDiscoveryFinished) {
+ this.partitions = new HashSet<>();
+ partitions.addAll(
+ assignPartitions.stream()
+ .map(
+ topicPartition ->
+ new TopicPartitionWithAssignStatus(
+ topicPartition,
+ TopicPartitionWithAssignStatus.ASSIGNED))
+ .collect(Collectors.toSet()));
+ partitions.addAll(
+ unAssignInitialPartitions.stream()
+ .map(
+ topicPartition ->
+ new TopicPartitionWithAssignStatus(
+ topicPartition,
+ TopicPartitionWithAssignStatus.UNASSIGNED_INITIAL))
+ .collect(Collectors.toSet()));
+ this.initialDiscoveryFinished = initialDiscoveryFinished;
+ }
+
+ public Set<TopicPartitionWithAssignStatus> partitions() {
+ return partitions;
}
public Set<TopicPartition> assignedPartitions() {
- return assignedPartitions;
+ return partitions.stream()
Review Comment:
Lines 68-74 and 78-84 duplicate the code a bit.
Maybe you can define a private method to abstract the common code and call it from `assignedPartitions()` and `unassignedPartitions()`
So, something like this
```
private Set<TopicPartition> filterPartitions(long assignmentStatus);
```
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -113,10 +124,13 @@ public KafkaSourceEnumerator(
Properties properties,
SplitEnumeratorContext<KafkaPartitionSplit> context,
Boundedness boundedness,
- Set<TopicPartition> assignedPartitions) {
+ Set<TopicPartition> assignedPartitions,
Review Comment:
It will be better to pass the `KafkaSourceEnumState` object in this constructor to limit the number of arguments.
--
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
[GitHub] [flink-connector-kafka] loserwang1024 commented on pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#issuecomment-1581782241
@PatrickRen CC
--
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
[GitHub] [flink-connector-kafka] loserwang1024 commented on a diff in pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#discussion_r1208763810
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumState.java:
##########
@@ -22,18 +22,69 @@
import org.apache.kafka.common.TopicPartition;
+import java.util.HashSet;
import java.util.Set;
+import java.util.stream.Collectors;
/** The state of Kafka source enumerator. */
@Internal
public class KafkaSourceEnumState {
- private final Set<TopicPartition> assignedPartitions;
+ /** Partitions with status: ASSIGNED or UNASSIGNED_INITIAL. */
+ private final Set<TopicPartitionWithAssignStatus> partitions;
+ /**
+ * this flag will be marked as true if inital partitions are discovered after enumerator starts.
+ */
+ private final boolean initialDiscoveryFinished;
- KafkaSourceEnumState(Set<TopicPartition> assignedPartitions) {
- this.assignedPartitions = assignedPartitions;
+ KafkaSourceEnumState(
+ Set<TopicPartition> assignPartitions,
+ Set<TopicPartition> unAssignInitialPartitions,
+ boolean initialDiscoveryFinished) {
+ this.partitions = new HashSet<>();
+ partitions.addAll(
+ assignPartitions.stream()
+ .map(
+ topicPartition ->
+ new TopicPartitionWithAssignStatus(
+ topicPartition,
+ TopicPartitionWithAssignStatus.ASSIGNED))
+ .collect(Collectors.toSet()));
+ partitions.addAll(
+ unAssignInitialPartitions.stream()
+ .map(
+ topicPartition ->
+ new TopicPartitionWithAssignStatus(
+ topicPartition,
+ TopicPartitionWithAssignStatus.UNASSIGNED_INITIAL))
+ .collect(Collectors.toSet()));
+ this.initialDiscoveryFinished = initialDiscoveryFinished;
+ }
+
+ public Set<TopicPartitionWithAssignStatus> partitions() {
+ return partitions;
}
public Set<TopicPartition> assignedPartitions() {
- return assignedPartitions;
+ return partitions.stream()
Review Comment:
@RamanVerma Thanks for your advice. Would you like to code review again?
--
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
[GitHub] [flink-connector-kafka] PatrickRen merged pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "PatrickRen (via GitHub)" <gi...@apache.org>.
PatrickRen merged PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28
--
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
[GitHub] [flink-connector-kafka] loserwang1024 closed pull request #28: EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 closed pull request #28: EARLIEST offset strategy for partitions discoveried later based on FLIP-288
URL: https://github.com/apache/flink-connector-kafka/pull/28
--
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
[GitHub] [flink-connector-kafka] loserwang1024 commented on pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#issuecomment-1562598385
@RamanVerma Thanks for your advice. Would you like to see my new modification?
--
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
[GitHub] [flink-connector-kafka] PatrickRen commented on a diff in pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "PatrickRen (via GitHub)" <gi...@apache.org>.
PatrickRen commented on code in PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#discussion_r1211150630
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumState.java:
##########
@@ -22,18 +22,67 @@
import org.apache.kafka.common.TopicPartition;
+import java.util.HashSet;
import java.util.Set;
+import java.util.stream.Collectors;
/** The state of Kafka source enumerator. */
@Internal
public class KafkaSourceEnumState {
- private final Set<TopicPartition> assignedPartitions;
+ /** Partitions with status: ASSIGNED or UNASSIGNED_INITIAL. */
+ private final Set<TopicPartitionAndAssignmentStatus> partitions;
+ /**
+ * this flag will be marked as true if inital partitions are discovered after enumerator starts.
+ */
+ private final boolean initialDiscoveryFinished;
- KafkaSourceEnumState(Set<TopicPartition> assignedPartitions) {
- this.assignedPartitions = assignedPartitions;
+ KafkaSourceEnumState(
+ Set<TopicPartition> assignPartitions,
+ Set<TopicPartition> unassignedInitialPartitions,
+ boolean initialDiscoveryFinished) {
+ this.partitions = new HashSet<>();
+ partitions.addAll(
+ assignPartitions.stream()
+ .map(
+ topicPartition ->
+ new TopicPartitionAndAssignmentStatus(
+ topicPartition,
+ TopicPartitionAndAssignmentStatus.ASSIGNED))
+ .collect(Collectors.toSet()));
+ partitions.addAll(
+ unassignedInitialPartitions.stream()
+ .map(
+ topicPartition ->
+ new TopicPartitionAndAssignmentStatus(
+ topicPartition,
+ TopicPartitionAndAssignmentStatus
+ .UNASSIGNED_INITIAL))
+ .collect(Collectors.toSet()));
+ this.initialDiscoveryFinished = initialDiscoveryFinished;
+ }
+
+ public Set<TopicPartitionAndAssignmentStatus> partitions() {
+ return partitions;
}
public Set<TopicPartition> assignedPartitions() {
- return assignedPartitions;
+ return filterPartitions(TopicPartitionAndAssignmentStatus.ASSIGNED);
+ }
+
+ public Set<TopicPartition> unassignedInitialPartitions() {
+ return filterPartitions(TopicPartitionAndAssignmentStatus.UNASSIGNED_INITIAL);
+ }
+
+ public boolean initialDiscoveryFinished() {
+ return initialDiscoveryFinished;
+ }
+
+ private Set<TopicPartition> filterPartitions(long assignmentStatus) {
Review Comment:
nit: Maybe use `filterPartitionsByAssignmentStatus` to make the semantic clearer.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/TopicPartitionAndAssignmentStatus.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.source.enumerator;
+
+import org.apache.flink.annotation.Internal;
+
+import org.apache.kafka.common.TopicPartition;
+
+/** Kafka partition with assign status. */
+@Internal
+public class TopicPartitionAndAssignmentStatus {
+ private final TopicPartition topicPartition;
+ private final long assignmentStatus;
+
+ /** Partitions that have been assigned to readers. */
+ public static final long ASSIGNED = -1;
Review Comment:
What about using an enum instead of magic numbers for `assignmentStatus`?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -56,54 +58,65 @@ public int getVersion() {
@Override
public byte[] serialize(KafkaSourceEnumState enumState) throws IOException {
- return serializeTopicPartitions(enumState.assignedPartitions());
- }
-
- @Override
- public KafkaSourceEnumState deserialize(int version, byte[] serialized) throws IOException {
- if (version == CURRENT_VERSION) {
- final Set<TopicPartition> assignedPartitions = deserializeTopicPartitions(serialized);
- return new KafkaSourceEnumState(assignedPartitions);
- }
-
- // Backward compatibility
- if (version == VERSION_0) {
- Map<Integer, Set<KafkaPartitionSplit>> currentPartitionAssignment =
- SerdeUtils.deserializeSplitAssignments(
- serialized, new KafkaPartitionSplitSerializer(), HashSet::new);
- Set<TopicPartition> currentAssignedSplits = new HashSet<>();
- currentPartitionAssignment.forEach(
- (reader, splits) ->
- splits.forEach(
- split -> currentAssignedSplits.add(split.getTopicPartition())));
- return new KafkaSourceEnumState(currentAssignedSplits);
- }
- throw new IOException(
- String.format(
- "The bytes are serialized with version %d, "
- + "while this deserializer only supports version up to %d",
- version, CURRENT_VERSION));
- }
-
- private static byte[] serializeTopicPartitions(Collection<TopicPartition> topicPartitions)
- throws IOException {
+ Set<TopicPartition> assignedPartitions = enumState.assignedPartitions();
+ Set<TopicPartition> unassignedInitialPartitions = enumState.unassignedInitialPartitions();
+ boolean initialDiscoveryFinished = enumState.initialDiscoveryFinished();
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream out = new DataOutputStream(baos)) {
- out.writeInt(topicPartitions.size());
- for (TopicPartition tp : topicPartitions) {
+ out.writeInt(assignedPartitions.size());
+ for (TopicPartition tp : assignedPartitions) {
out.writeUTF(tp.topic());
out.writeInt(tp.partition());
}
out.flush();
Review Comment:
What about wrapping these lines into a helper function, like `serializeTopicPartitions`? These lines are copy-pasted below for serializing `unassignedInitialPartitions`
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -266,10 +286,18 @@ private void checkPartitionChanges(Set<TopicPartition> fetchedPartitions, Throwa
private PartitionSplitChange initializePartitionSplits(PartitionChange partitionChange) {
Set<TopicPartition> newPartitions =
Collections.unmodifiableSet(partitionChange.getNewPartitions());
+
OffsetsInitializer.PartitionOffsetsRetriever offsetsRetriever = getOffsetsRetriever();
+ // intial partitions use OffsetsInitializer specified by the user while new partitions use
Review Comment:
nit: `intial` -> `initial`
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java:
##########
@@ -632,6 +702,31 @@ private void verifySplitAssignmentWithPartitions(
assertThat(actualTopicPartitions).isEqualTo(allTopicPartitionsFromAssignment);
}
+ /**
+ * get all assigned partition splits of topics.
+ *
+ * @param context
+ * @param topics
+ * @return
Review Comment:
Uncompleted JavaDocs
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -121,4 +134,54 @@ private static Set<TopicPartition> deserializeTopicPartitions(byte[] serializedT
return topicPartitions;
}
}
+
+ private static KafkaSourceEnumState deserializeTopicPartitionsV2(
Review Comment:
`deserializeEnumStateV2(byte[] serializedEnumState)`. This function deserializes the entire enum state instead of `TopicPartition`s.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -46,8 +47,9 @@
private static final int VERSION_0 = 0;
private static final int VERSION_1 = 1;
+ private static final int VERSION_2 = 2;
Review Comment:
Adding some comments describing the format of each version will help maintaining the code in the future.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializerTest.java:
##########
@@ -90,6 +106,23 @@ private Set<TopicPartition> constructTopicPartitions() {
return topicPartitions;
}
+ private Set<TopicPartition> constructNewTopicPartitions() {
+ // 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.
+ // Totally NUM_READERS * NUM_PARTITIONS_PER_TOPIC partitions will be created.
Review Comment:
These comments doesn't match the logic below.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -121,4 +134,54 @@ private static Set<TopicPartition> deserializeTopicPartitions(byte[] serializedT
return topicPartitions;
}
}
+
+ private static KafkaSourceEnumState deserializeTopicPartitionsV2(
+ byte[] serializedTopicPartitions) throws IOException {
+
+ try (ByteArrayInputStream bais = new ByteArrayInputStream(serializedTopicPartitions);
+ DataInputStream in = new DataInputStream(bais)) {
+
+ final int numAssignedPartitions = in.readInt();
+ Set<TopicPartition> assignedPartitions = new HashSet<>(numAssignedPartitions);
+ for (int i = 0; i < numAssignedPartitions; i++) {
+ final String topic = in.readUTF();
+ final int partition = in.readInt();
+ assignedPartitions.add(new TopicPartition(topic, partition));
+ }
Review Comment:
These lines should also be wrapped into a helper function `deserializeTopicPartitions`, as they are copy-pasted again below.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -56,54 +58,65 @@ public int getVersion() {
@Override
public byte[] serialize(KafkaSourceEnumState enumState) throws IOException {
- return serializeTopicPartitions(enumState.assignedPartitions());
- }
-
- @Override
- public KafkaSourceEnumState deserialize(int version, byte[] serialized) throws IOException {
- if (version == CURRENT_VERSION) {
- final Set<TopicPartition> assignedPartitions = deserializeTopicPartitions(serialized);
- return new KafkaSourceEnumState(assignedPartitions);
- }
-
- // Backward compatibility
- if (version == VERSION_0) {
- Map<Integer, Set<KafkaPartitionSplit>> currentPartitionAssignment =
- SerdeUtils.deserializeSplitAssignments(
- serialized, new KafkaPartitionSplitSerializer(), HashSet::new);
- Set<TopicPartition> currentAssignedSplits = new HashSet<>();
- currentPartitionAssignment.forEach(
- (reader, splits) ->
- splits.forEach(
- split -> currentAssignedSplits.add(split.getTopicPartition())));
- return new KafkaSourceEnumState(currentAssignedSplits);
- }
- throw new IOException(
- String.format(
- "The bytes are serialized with version %d, "
- + "while this deserializer only supports version up to %d",
- version, CURRENT_VERSION));
- }
-
- private static byte[] serializeTopicPartitions(Collection<TopicPartition> topicPartitions)
- throws IOException {
+ Set<TopicPartition> assignedPartitions = enumState.assignedPartitions();
+ Set<TopicPartition> unassignedInitialPartitions = enumState.unassignedInitialPartitions();
+ boolean initialDiscoveryFinished = enumState.initialDiscoveryFinished();
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream out = new DataOutputStream(baos)) {
- out.writeInt(topicPartitions.size());
- for (TopicPartition tp : topicPartitions) {
+ out.writeInt(assignedPartitions.size());
+ for (TopicPartition tp : assignedPartitions) {
out.writeUTF(tp.topic());
out.writeInt(tp.partition());
}
out.flush();
+ out.writeInt(unassignedInitialPartitions.size());
+ for (TopicPartition tp : unassignedInitialPartitions) {
+ out.writeUTF(tp.topic());
+ out.writeInt(tp.partition());
+ }
+ out.flush();
+
+ out.writeBoolean(initialDiscoveryFinished);
+
return baos.toByteArray();
}
}
- private static Set<TopicPartition> deserializeTopicPartitions(byte[] serializedTopicPartitions)
- throws IOException {
+ @Override
+ public KafkaSourceEnumState deserialize(int version, byte[] serialized) throws IOException {
+ switch (version) {
+ case CURRENT_VERSION:
+ return deserializeTopicPartitionsV2(serialized);
+ case VERSION_1:
+ final Set<TopicPartition> assignedPartitions =
+ deserializeTopicPartitionsV1(serialized);
+ return new KafkaSourceEnumState(assignedPartitions, new HashSet<>(), true);
+ case VERSION_0:
+ Map<Integer, Set<KafkaPartitionSplit>> currentPartitionAssignment =
+ SerdeUtils.deserializeSplitAssignments(
+ serialized, new KafkaPartitionSplitSerializer(), HashSet::new);
+ Set<TopicPartition> currentAssignedSplits = new HashSet<>();
+ currentPartitionAssignment.forEach(
+ (reader, splits) ->
+ splits.forEach(
+ split ->
+ currentAssignedSplits.add(
+ split.getTopicPartition())));
+ return new KafkaSourceEnumState(currentAssignedSplits, new HashSet<>(), true);
+ default:
+ throw new IOException(
+ String.format(
+ "The bytes are serialized with version %d, "
+ + "while this deserializer only supports version up to %d",
+ version, CURRENT_VERSION));
+ }
+ }
+
+ private static Set<TopicPartition> deserializeTopicPartitionsV1(
Review Comment:
This is actually the `deserializeTopicPartitions`. It is the serialized enum state that has different versions, instead of `TopicPartition`.
--
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
[GitHub] [flink-connector-kafka] PatrickRen commented on a diff in pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "PatrickRen (via GitHub)" <gi...@apache.org>.
PatrickRen commented on code in PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#discussion_r1244818295
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -56,69 +67,114 @@ public int getVersion() {
@Override
public byte[] serialize(KafkaSourceEnumState enumState) throws IOException {
- return serializeTopicPartitions(enumState.assignedPartitions());
+ Set<TopicPartitionAndAssignmentStatus> partitions = enumState.partitions();
+ boolean initialDiscoveryFinished = enumState.initialDiscoveryFinished();
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+ out.writeInt(partitions.size());
+ for (TopicPartitionAndAssignmentStatus topicPartitionAndAssignmentStatus : partitions) {
+ out.writeUTF(topicPartitionAndAssignmentStatus.topicPartition().topic());
+ out.writeInt(topicPartitionAndAssignmentStatus.topicPartition().partition());
+ out.writeInt(topicPartitionAndAssignmentStatus.assignmentStatus().getStatusCode());
+ }
+ out.flush();
+ out.writeBoolean(initialDiscoveryFinished);
+ out.flush();
Review Comment:
Is there any specific reason to flush twice here?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/AssignmentStatus.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.source.enumerator;
+
+/** status of partition assignment. */
+public enum AssignmentStatus {
+
+ /** Partitions that have been assigned to readers. */
+ ASSIGNED(0),
+ /**
+ * The partitions that have been discovered during initialization but not assigned to readers
+ * yet.
+ */
+ UNASSIGNED_INITIAL(1);
+ private int statusCode;
Review Comment:
private **final** int statusCode
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -88,6 +95,8 @@ public class KafkaSourceEnumerator
// This flag will be marked as true if periodically partition discovery is disabled AND the
// initializing partition discovery has finished.
private boolean noMoreNewPartitionSplits = false;
+ // this flag will be marked as true if inital partitions are discovered after enumerator starts
Review Comment:
typo: `inital` -> `initial`
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/AssignmentStatus.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.source.enumerator;
+
+/** status of partition assignment. */
+public enum AssignmentStatus {
Review Comment:
We need to mark this new class as `@Internal`
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -56,69 +67,114 @@ public int getVersion() {
@Override
public byte[] serialize(KafkaSourceEnumState enumState) throws IOException {
- return serializeTopicPartitions(enumState.assignedPartitions());
+ Set<TopicPartitionAndAssignmentStatus> partitions = enumState.partitions();
+ boolean initialDiscoveryFinished = enumState.initialDiscoveryFinished();
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+ out.writeInt(partitions.size());
+ for (TopicPartitionAndAssignmentStatus topicPartitionAndAssignmentStatus : partitions) {
+ out.writeUTF(topicPartitionAndAssignmentStatus.topicPartition().topic());
+ out.writeInt(topicPartitionAndAssignmentStatus.topicPartition().partition());
+ out.writeInt(topicPartitionAndAssignmentStatus.assignmentStatus().getStatusCode());
+ }
+ out.flush();
+ out.writeBoolean(initialDiscoveryFinished);
+ out.flush();
+ return baos.toByteArray();
+ }
}
@Override
public KafkaSourceEnumState deserialize(int version, byte[] serialized) throws IOException {
- if (version == CURRENT_VERSION) {
- final Set<TopicPartition> assignedPartitions = deserializeTopicPartitions(serialized);
- return new KafkaSourceEnumState(assignedPartitions);
+ switch (version) {
+ case CURRENT_VERSION:
+ return deserializeTopicPartitionAndAssignmentStatus(serialized);
+ case VERSION_1:
+ final Set<TopicPartition> assignedPartitions =
+ deserializeTopicPartitions(serialized);
+ return new KafkaSourceEnumState(assignedPartitions, new HashSet<>(), true);
+ case VERSION_0:
+ Map<Integer, Set<KafkaPartitionSplit>> currentPartitionAssignment =
+ SerdeUtils.deserializeSplitAssignments(
+ serialized, new KafkaPartitionSplitSerializer(), HashSet::new);
+ Set<TopicPartition> currentAssignedSplits = new HashSet<>();
+ currentPartitionAssignment.forEach(
+ (reader, splits) ->
+ splits.forEach(
+ split ->
+ currentAssignedSplits.add(
+ split.getTopicPartition())));
+ return new KafkaSourceEnumState(currentAssignedSplits, new HashSet<>(), true);
+ default:
+ throw new IOException(
+ String.format(
+ "The bytes are serialized with version %d, "
+ + "while this deserializer only supports version up to %d",
+ version, CURRENT_VERSION));
}
-
- // Backward compatibility
- if (version == VERSION_0) {
- Map<Integer, Set<KafkaPartitionSplit>> currentPartitionAssignment =
- SerdeUtils.deserializeSplitAssignments(
- serialized, new KafkaPartitionSplitSerializer(), HashSet::new);
- Set<TopicPartition> currentAssignedSplits = new HashSet<>();
- currentPartitionAssignment.forEach(
- (reader, splits) ->
- splits.forEach(
- split -> currentAssignedSplits.add(split.getTopicPartition())));
- return new KafkaSourceEnumState(currentAssignedSplits);
- }
-
- throw new IOException(
- String.format(
- "The bytes are serialized with version %d, "
- + "while this deserializer only supports version up to %d",
- version, CURRENT_VERSION));
}
- private static byte[] serializeTopicPartitions(Collection<TopicPartition> topicPartitions)
+ private static Set<TopicPartition> deserializeTopicPartitions(byte[] serializedTopicPartitions)
throws IOException {
- try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
- DataOutputStream out = new DataOutputStream(baos)) {
+ try (ByteArrayInputStream bais = new ByteArrayInputStream(serializedTopicPartitions);
+ DataInputStream in = new DataInputStream(bais)) {
- out.writeInt(topicPartitions.size());
- for (TopicPartition tp : topicPartitions) {
- out.writeUTF(tp.topic());
- out.writeInt(tp.partition());
+ final int numPartitions = in.readInt();
+ Set<TopicPartition> topicPartitions = new HashSet<>(numPartitions);
+ for (int i = 0; i < numPartitions; i++) {
+ final String topic = in.readUTF();
+ final int partition = in.readInt();
+ topicPartitions.add(new TopicPartition(topic, partition));
+ }
+ if (in.available() > 0) {
+ throw new IOException("Unexpected trailing bytes in serialized topic partitions");
}
- out.flush();
- return baos.toByteArray();
+ return topicPartitions;
}
}
- private static Set<TopicPartition> deserializeTopicPartitions(byte[] serializedTopicPartitions)
- throws IOException {
+ private static KafkaSourceEnumState deserializeTopicPartitionAndAssignmentStatus(
+ byte[] serializedTopicPartitions) throws IOException {
Review Comment:
The name of the argument is not quite appropriate. These bytes are actually the entire serialized `KafkaSourceEnumState` instead of just `TopicPartition`s.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializerTest.java:
##########
@@ -53,37 +57,53 @@ public void testEnumStateSerde() throws IOException {
serializer.deserialize(serializer.getVersion(), bytes);
assertThat(restoredState.assignedPartitions()).isEqualTo(state.assignedPartitions());
+ assertThat(restoredState.unassignedInitialPartitions())
+ .isEqualTo(state.unassignedInitialPartitions());
+ assertThat(restoredState.initialDiscoveryFinished()).isTrue();
}
@Test
public void testBackwardCompatibility() throws IOException {
- final Set<TopicPartition> topicPartitions = constructTopicPartitions();
+ final Set<TopicPartition> topicPartitions = constructTopicPartitions(0);
final Map<Integer, Set<KafkaPartitionSplit>> splitAssignments =
toSplitAssignments(topicPartitions);
// Create bytes in the way of KafkaEnumStateSerializer version 0 doing serialization
- final byte[] bytes =
+ 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);
- // Deserialize above bytes with KafkaEnumStateSerializer version 1 to check backward
+ // Deserialize above bytes with KafkaEnumStateSerializer version 2 to check backward
// compatibility
- final KafkaSourceEnumState kafkaSourceEnumState =
- new KafkaSourceEnumStateSerializer().deserialize(0, bytes);
-
- assertThat(kafkaSourceEnumState.assignedPartitions()).isEqualTo(topicPartitions);
+ final KafkaSourceEnumState kafkaSourceEnumStateV0 =
+ new KafkaSourceEnumStateSerializer().deserialize(0, bytesV0);
+ final KafkaSourceEnumState kafkaSourceEnumStateV1 =
+ new KafkaSourceEnumStateSerializer().deserialize(1, bytesV1);
+
+ assertThat(kafkaSourceEnumStateV0.assignedPartitions()).isEqualTo(topicPartitions);
+ assertThat(kafkaSourceEnumStateV0.unassignedInitialPartitions()).isEmpty();
+ assertThat(kafkaSourceEnumStateV0.initialDiscoveryFinished()).isTrue();
+
+ assertThat(kafkaSourceEnumStateV1.assignedPartitions()).isEqualTo(topicPartitions);
+ assertThat(kafkaSourceEnumStateV1.unassignedInitialPartitions()).isEmpty();
+ assertThat(kafkaSourceEnumStateV1.initialDiscoveryFinished()).isTrue();
}
- private Set<TopicPartition> constructTopicPartitions() {
+ private Set<TopicPartition> constructTopicPartitions(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.
+ // NUM_PARTITIONS_PER_TOPIC partitions, which is start from startPartition.
Review Comment:
NUM_PARTITIONS_PER_TOPIC partitions ~~, which is start from startPartition.~~ The starting partition number is `startPartition`.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/AssignmentStatus.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.source.enumerator;
+
+/** status of partition assignment. */
+public enum AssignmentStatus {
+
+ /** Partitions that have been assigned to readers. */
+ ASSIGNED(0),
+ /**
+ * The partitions that have been discovered during initialization but not assigned to readers
+ * yet.
+ */
+ UNASSIGNED_INITIAL(1);
+ private int statusCode;
+
+ AssignmentStatus(int statusCode) {
+ this.statusCode = statusCode;
+ }
+
+ public int getStatusCode() {
+ return statusCode;
+ }
+
+ public static AssignmentStatus findEnumByCode(int statusCode) {
Review Comment:
Maybe use `ofStatusCode` as the name of the function?
--
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
[GitHub] [flink-connector-kafka] boring-cyborg[bot] commented on pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#issuecomment-1624546827
Awesome work, congrats on your first merged pull request!
--
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
[GitHub] [flink-connector-kafka] RamanVerma commented on a diff in pull request #28: [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288
Posted by "RamanVerma (via GitHub)" <gi...@apache.org>.
RamanVerma commented on code in PR #28:
URL: https://github.com/apache/flink-connector-kafka/pull/28#discussion_r1190320189
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -56,54 +58,75 @@ public int getVersion() {
@Override
public byte[] serialize(KafkaSourceEnumState enumState) throws IOException {
- return serializeTopicPartitions(enumState.assignedPartitions());
+
+ Set<TopicPartition> assignedPartitions = enumState.assignedPartitions();
+ Set<TopicPartition> unassignedInitialPartitons = enumState.unassignedInitialPartitons();
+ boolean initialDiscoveryFinished = enumState.initialDiscoveryFinished();
+ return serializeTopicPartitions(
+ assignedPartitions, unassignedInitialPartitons, initialDiscoveryFinished);
}
@Override
public KafkaSourceEnumState deserialize(int version, byte[] serialized) throws IOException {
- if (version == CURRENT_VERSION) {
- final Set<TopicPartition> assignedPartitions = deserializeTopicPartitions(serialized);
- return new KafkaSourceEnumState(assignedPartitions);
- }
-
- // Backward compatibility
- if (version == VERSION_0) {
- Map<Integer, Set<KafkaPartitionSplit>> currentPartitionAssignment =
- SerdeUtils.deserializeSplitAssignments(
- serialized, new KafkaPartitionSplitSerializer(), HashSet::new);
- Set<TopicPartition> currentAssignedSplits = new HashSet<>();
- currentPartitionAssignment.forEach(
- (reader, splits) ->
- splits.forEach(
- split -> currentAssignedSplits.add(split.getTopicPartition())));
- return new KafkaSourceEnumState(currentAssignedSplits);
+ switch (version) {
+ case CURRENT_VERSION:
+ return deserializeTopicPartitionsV2(serialized);
+ case VERSION_1:
+ final Set<TopicPartition> assignedPartitions =
+ deserializeTopicPartitionsV1(serialized);
+ return new KafkaSourceEnumState(assignedPartitions, new HashSet<>(), true);
+ case VERSION_0:
+ Map<Integer, Set<KafkaPartitionSplit>> currentPartitionAssignment =
+ SerdeUtils.deserializeSplitAssignments(
+ serialized, new KafkaPartitionSplitSerializer(), HashSet::new);
+ Set<TopicPartition> currentAssignedSplits = new HashSet<>();
+ currentPartitionAssignment.forEach(
+ (reader, splits) ->
+ splits.forEach(
+ split ->
+ currentAssignedSplits.add(
+ split.getTopicPartition())));
+ return new KafkaSourceEnumState(currentAssignedSplits, new HashSet<>(), true);
+ default:
+ throw new IOException(
+ String.format(
+ "The bytes are serialized with version %d, "
+ + "while this deserializer only supports version up to %d",
+ version, CURRENT_VERSION));
}
-
- throw new IOException(
- String.format(
- "The bytes are serialized with version %d, "
- + "while this deserializer only supports version up to %d",
- version, CURRENT_VERSION));
}
- private static byte[] serializeTopicPartitions(Collection<TopicPartition> topicPartitions)
+ private static byte[] serializeTopicPartitions(
+ Collection<TopicPartition> assignedPartitions,
+ Collection<TopicPartition> unassignedInitialPartitons,
Review Comment:
typo `unassignedInitialPartitons` -> `unassignedInitialPartitions`
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -56,54 +58,75 @@ public int getVersion() {
@Override
public byte[] serialize(KafkaSourceEnumState enumState) throws IOException {
- return serializeTopicPartitions(enumState.assignedPartitions());
+
+ Set<TopicPartition> assignedPartitions = enumState.assignedPartitions();
+ Set<TopicPartition> unassignedInitialPartitons = enumState.unassignedInitialPartitons();
+ boolean initialDiscoveryFinished = enumState.initialDiscoveryFinished();
+ return serializeTopicPartitions(
Review Comment:
Maybe we can just get rid of the private method now.
We are serializing more than just topic partitions (initialDiscoveryFinished is a boolean) so the method name needs to change. Also, there is no other caller. So, let's just do everything in serialize method itself.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java:
##########
@@ -121,4 +144,54 @@ private static Set<TopicPartition> deserializeTopicPartitions(byte[] serializedT
return topicPartitions;
}
}
+
+ private static KafkaSourceEnumState deserializeTopicPartitionsV2(
+ byte[] serializedTopicPartitions) throws IOException {
+
+ try (ByteArrayInputStream bais = new ByteArrayInputStream(serializedTopicPartitions);
+ DataInputStream in = new DataInputStream(bais)) {
+
+ final int numAssignedPartitions = in.readInt();
+ Set<TopicPartition> assignedPartitions = new HashSet<>(numAssignedPartitions);
+ for (int i = 0; i < numAssignedPartitions; i++) {
+ final String topic = in.readUTF();
+ final int partition = in.readInt();
+ assignedPartitions.add(new TopicPartition(topic, partition));
+ }
+ final int numUnassignedInitialPartitons = in.readInt();
Review Comment:
typos: `numUnassignedInitialPartitons`. Also in line 162
--
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