[FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288 #28

loserwang1024 · 2023-05-08T04:12:35Z

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), 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

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.
Add a newDiscoveryOffsetsInitializer(EARLIEST) to get offsets for newly discovered partitions.
Modify kafkaSourceEnumStateSerializer to handle the expanded KafkaSourceEnumState.

Verifying this change

Test the backward compatibility of state when deserializing in KafkaSourceEnumStateSerializerTest.
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
Expand KafkaEnumeratorTest#testDiscoverPartitionsPeriodically method to test whether new partitions use EARLIEST offset while initial partitions use specified offset strategy.

RamanVerma

Thanks for the changes @loserwang1024
I have left some initial comments.

RamanVerma · 2023-05-10T19:00:49Z

...a/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumState.java

-    KafkaSourceEnumState(Set<TopicPartition> assignedPartitions) {
-        this.assignedPartitions = assignedPartitions;
+    KafkaSourceEnumState(
+            Set<TopicPartition> assignPartitions,


please change the parameter names to assignedPartitions and unassignedInitialPartitions

RamanVerma · 2023-05-10T19:08:21Z

.../java/org/apache/flink/connector/kafka/source/enumerator/TopicPartitionWithAssignStatus.java

+@Internal
+public class TopicPartitionWithAssignStatus {
+    private final TopicPartition topicPartition;
+    private final long assignStatus;


assignmentStatus would convey the meaning better than assignStatus
Also, I would prefer TopicPartitionAndAssignmentStatus over TopicPartitionWithAssignStatus

RamanVerma · 2023-05-10T19:14:39Z

...a/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumState.java

    }

    public Set<TopicPartition> assignedPartitions() {
-        return assignedPartitions;
+        return partitions.stream()


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);

@RamanVerma Thanks for your advice. Would you like to code review again?

RamanVerma · 2023-05-10T19:17:43Z

.../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,


It will be better to pass the KafkaSourceEnumState object in this constructor to limit the number of arguments.

RamanVerma · 2023-05-10T19:34:42Z

.../java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java

-    private static byte[] serializeTopicPartitions(Collection<TopicPartition> topicPartitions)
+    private static byte[] serializeTopicPartitions(
+            Collection<TopicPartition> assignedPartitions,
+            Collection<TopicPartition> unassignedInitialPartitons,


typo unassignedInitialPartitons -> unassignedInitialPartitions

RamanVerma · 2023-05-10T19:41:35Z

.../java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java

+        Set<TopicPartition> assignedPartitions = enumState.assignedPartitions();
+        Set<TopicPartition> unassignedInitialPartitons = enumState.unassignedInitialPartitons();
+        boolean initialDiscoveryFinished = enumState.initialDiscoveryFinished();
+        return serializeTopicPartitions(


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.

RamanVerma · 2023-05-10T19:52:42Z

.../java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java

+                final int partition = in.readInt();
+                assignedPartitions.add(new TopicPartition(topic, partition));
+            }
+            final int numUnassignedInitialPartitons = in.readInt();


typos: numUnassignedInitialPartitons. Also in line 162

loserwang1024 · 2023-05-25T09:39:19Z

@RamanVerma Thanks for your advice. Would you like to see my new modification?

loserwang1024 · 2023-05-30T02:20:04Z

Now that @RamanVerma is busy, could anyone else help me? CC, @PatrickRen

PatrickRen

@loserwang1024 Thanks for the PR! The overall logic looks good to me. I left some comments about naming and code style issues.

...a/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumState.java

.../java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java

...va/org/apache/flink/connector/kafka/source/enumerator/TopicPartitionAndAssignmentStatus.java

.../java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java

...a/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializerTest.java

.../src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java

...ka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java

loserwang1024 · 2023-06-08T02:07:20Z

@PatrickRen CC

PatrickRen

@loserwang1024 Thanks for the update! I left some comments.

.../src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java

.../java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java

...kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/AssignmentStatus.java

.../java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializer.java

...a/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumStateSerializerTest.java

PatrickRen

@loserwang1024 Thanks for the update. LGTM.

Could you squash all commits into one and rebase the latest master? I triggered a CI just now and let's wait for the result.

…s discoveried later based on FLIP-288

boring-cyborg · 2023-07-07T02:19:12Z

Awesome work, congrats on your first merged pull request!

boring-cyborg bot added the component=Connectors/Kafka label May 8, 2023

loserwang1024 closed this May 8, 2023

loserwang1024 changed the title ~~EARLIEST offset strategy for partitions discoveried later based on FLIP-288~~ [FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288 May 8, 2023

loserwang1024 reopened this May 8, 2023

RamanVerma reviewed May 10, 2023

View reviewed changes

loserwang1024 force-pushed the FLIP288-FLINK-32019 branch 2 times, most recently from eec6121 to 56c87e0 Compare May 23, 2023 08:48

loserwang1024 requested a review from RamanVerma May 23, 2023 08:50

PatrickRen requested changes May 31, 2023

View reviewed changes

PatrickRen self-assigned this Jun 1, 2023

loserwang1024 force-pushed the FLIP288-FLINK-32019 branch from 56c87e0 to 6d56e91 Compare June 7, 2023 02:58

loserwang1024 requested a review from PatrickRen June 7, 2023 06:13

PatrickRen reviewed Jun 28, 2023

View reviewed changes

PatrickRen approved these changes Jun 29, 2023

View reviewed changes

[FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partition…

07c3165

…s discoveried later based on FLIP-288

loserwang1024 force-pushed the FLIP288-FLINK-32019 branch from deae500 to 07c3165 Compare June 30, 2023 09:14

PatrickRen merged commit ad62c13 into apache:main Jul 7, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288 #28

[FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288 #28

loserwang1024 commented May 8, 2023

RamanVerma left a comment

RamanVerma May 10, 2023

RamanVerma May 10, 2023

RamanVerma May 10, 2023

loserwang1024 May 29, 2023

RamanVerma May 10, 2023

RamanVerma May 10, 2023

RamanVerma May 10, 2023

RamanVerma May 10, 2023

loserwang1024 commented May 25, 2023

loserwang1024 commented May 30, 2023

PatrickRen left a comment

loserwang1024 commented Jun 8, 2023

PatrickRen left a comment

PatrickRen left a comment •

edited

Loading

boring-cyborg bot commented Jul 7, 2023

[FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288 #28

[FLINK-32019][Connector/Kafka] EARLIEST offset strategy for partitions discoveried later based on FLIP-288 #28

Conversation

loserwang1024 commented May 8, 2023

What is the purpose of the change

Brief change log

Verifying this change

RamanVerma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

loserwang1024 commented May 25, 2023

loserwang1024 commented May 30, 2023

PatrickRen left a comment

Choose a reason for hiding this comment

loserwang1024 commented Jun 8, 2023

PatrickRen left a comment

Choose a reason for hiding this comment

PatrickRen left a comment • edited Loading

Choose a reason for hiding this comment

boring-cyborg bot commented Jul 7, 2023

PatrickRen left a comment •

edited

Loading