[FLINK-35228][Connectors/Kafka] Fix: DynamicKafkaSource does not read re-added topic for the same cluster #97

IgnasD · 2024-04-24T08:49:40Z

Saw some dormant code that was collecting activeTopicPartitions which got used nowhere. I believe the cleanup was already intended by the original developer, but just got lost for some reason.

So I've finished up the cleanup logic and added a test case to cover the issue in question.

boring-cyborg · 2024-04-24T08:49:44Z

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

IgnasD · 2024-04-24T08:55:12Z

Sorry, I forgot to change issue number while copying from another PR as a template. Fixed now.

mas-chen

Thanks for the contribution! The issue you are seeing and integration test make sense to me but I don't understand how you have fixed it. The changes to DynamicKafkaSourceEnumerator looks like solely refactoring. Did you forget to commit the fix?

IgnasD · 2024-04-25T06:46:07Z

The fix is that previously activeTopicPartitions were being constructed but used nowhere. Now I'm constructing KafkaSourceEnumState and passing activeTopicPartitions to it.

And yes, while doing that some refactoring came in naturally to avoid code duplication.

mas-chen · 2024-04-26T02:28:33Z

Thanks for explaining, I missed that! Rerunning CI, the PR LGTM

...org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java

mas-chen

left one comment. Can you also fixup the PR title? e.g. your jira ticket title is a bit more informative: "DynamicKafkaSource does not read re-added topic for the same cluster"

IgnasD · 2024-04-26T14:18:03Z

Changed the title as requested. Also, I've added filtering to unassignedInitialPartitions as suggested and covered it with a test case.

...apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java

mas-chen · 2024-04-27T19:38:44Z

Rerunning CI--seems to be flaky behavior in IT tests:

2024-04-27T19:26:47.5915234Z 19:26:47,500 [    Flat Map (1/1)#0] WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Flat Map (1/1)#0 (ae4b22de7586eaf3e02a3d7930e9290c_0a448493b4782967b150582570326227_0_0) switched from RUNNING to FAILED with failure cause:
2024-04-27T19:26:47.5917050Z org.apache.flink.test.util.SuccessException: null
2024-04-27T19:26:47.5918591Z 	at org.apache.flink.streaming.connectors.kafka.KafkaConsumerTestBase$10.flatMap(KafkaConsumerTestBase.java:1331) ~[test-classes/:?]
2024-04-27T19:26:47.5973189Z 	at org.apache.flink.streaming.connectors.kafka.KafkaConsumerTestBase$10.flatMap(KafkaConsumerTestBase.java:1302) ~[test-classes/:?]
2024-04-27T19:26:47.5975836Z 	at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:47) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.5978671Z 	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:237) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.5982139Z 	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:146) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.5985635Z 	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.5988935Z 	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.5991838Z 	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:562) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.5994539Z 	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.5997203Z 	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:858) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.5999551Z 	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:807) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6001791Z 	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953) ~[flink-runtime-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6003944Z 	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932) [flink-runtime-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6005833Z 	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746) [flink-runtime-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6007601Z 	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) [flink-runtime-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6008802Z 	at java.lang.Thread.run(Thread.java:840) [?:?]
2024-04-27T19:26:47.6011187Z 19:26:47,500 [    Flat Map (1/1)#0] INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Freeing task resources for Flat Map (1/1)#0 (ae4b22de7586eaf3e02a3d7930e9290c_0a448493b4782967b150582570326227_0_0).
2024-04-27T19:26:47.6015163Z 19:26:47,501 [flink-pekko.actor.default-dispatcher-9] INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Un-registering task and sending final execution state FAILED to JobManager for task Flat Map (1/1)#0 ae4b22de7586eaf3e02a3d7930e9290c_0a448493b4782967b150582570326227_0_0.
2024-04-27T19:26:47.6020124Z 19:26:47,501 [flink-pekko.actor.default-dispatcher-9] INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Flat Map (1/1) (ae4b22de7586eaf3e02a3d7930e9290c_0a448493b4782967b150582570326227_0_0) switched from RUNNING to FAILED on b7795d57-330f-46e5-80b4-9fed34414ad2 @ localhost (dataPort=-1).
2024-04-27T19:26:47.6022736Z org.apache.flink.test.util.SuccessException: null
2024-04-27T19:26:47.6024360Z 	at org.apache.flink.streaming.connectors.kafka.KafkaConsumerTestBase$10.flatMap(KafkaConsumerTestBase.java:1331) ~[test-classes/:?]
2024-04-27T19:26:47.6026692Z 	at org.apache.flink.streaming.connectors.kafka.KafkaConsumerTestBase$10.flatMap(KafkaConsumerTestBase.java:1302) ~[test-classes/:?]
2024-04-27T19:26:47.6096328Z 	at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:47) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6101993Z 	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:237) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6105266Z 	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:146) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6108323Z 	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6111209Z 	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6113917Z 	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:562) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6116391Z 	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6118929Z 	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:858) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6121163Z 	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:807) ~[flink-streaming-java-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6123314Z 	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953) ~[flink-runtime-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6125337Z 	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932) ~[flink-runtime-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6127149Z 	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746) ~[flink-runtime-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6129175Z 	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) ~[flink-runtime-1.18.1.jar:1.18.1]
2024-04-27T19:26:47.6130366Z 	at java.lang.Thread.run(Thread.java:840) ~[?:?]
2024-04-27T19:26:47.6343309Z 19:26:47,502 [flink-pekko.actor.default-dispatcher-9] INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Job Count elements from the topics (eccfd8eb6c25a421fae454d2b87488f3) switched from state RUNNING to FAILING.
2024-04-27T19:26:47.6356058Z org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy

...apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java

… re-added topic for the same cluster

mas-chen

Thanks for the iterations! LGTM

boring-cyborg · 2024-04-29T23:13:39Z

Awesome work, congrats on your first merged pull request!

boring-cyborg bot added the component=Connectors/Kafka label Apr 24, 2024

IgnasD changed the title ~~[FLINK-35138][Connectors/Kafka] Fix DynamicKafkaSourceEnumerator removed topics state cleanup~~ [FLINK-35228][Connectors/Kafka] Fix DynamicKafkaSourceEnumerator removed topics state cleanup Apr 24, 2024

IgnasD force-pushed the flink-35228 branch from 74ab28c to 6becb86 Compare April 24, 2024 08:54

MartijnVisser requested a review from mas-chen April 24, 2024 21:14

mas-chen reviewed Apr 24, 2024

View reviewed changes

mas-chen reviewed Apr 26, 2024

View reviewed changes

...org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java Outdated Show resolved Hide resolved

mas-chen reviewed Apr 26, 2024

View reviewed changes

IgnasD force-pushed the flink-35228 branch from 6becb86 to cbf3216 Compare April 26, 2024 14:15

IgnasD changed the title ~~[FLINK-35228][Connectors/Kafka] Fix DynamicKafkaSourceEnumerator removed topics state cleanup~~ [FLINK-35228][Connectors/Kafka] Fix: DynamicKafkaSource does not read re-added topic for the same cluster Apr 26, 2024

mas-chen reviewed Apr 26, 2024

View reviewed changes

...apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java Outdated Show resolved Hide resolved

IgnasD force-pushed the flink-35228 branch from cbf3216 to 0b418eb Compare April 27, 2024 15:08

mas-chen reviewed Apr 28, 2024

View reviewed changes

...apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java Outdated Show resolved Hide resolved

[FLINK-35228][Connectors/Kafka] Fix: DynamicKafkaSource does not read…

299b28f

… re-added topic for the same cluster

IgnasD force-pushed the flink-35228 branch from 0b418eb to 299b28f Compare April 29, 2024 09:01

mas-chen approved these changes Apr 29, 2024

View reviewed changes

mas-chen merged commit 00c9c8c into apache:main Apr 29, 2024
13 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-35228][Connectors/Kafka] Fix: DynamicKafkaSource does not read re-added topic for the same cluster #97

[FLINK-35228][Connectors/Kafka] Fix: DynamicKafkaSource does not read re-added topic for the same cluster #97

IgnasD commented Apr 24, 2024

boring-cyborg bot commented Apr 24, 2024

IgnasD commented Apr 24, 2024

mas-chen left a comment

IgnasD commented Apr 25, 2024

mas-chen commented Apr 26, 2024

mas-chen left a comment

IgnasD commented Apr 26, 2024

mas-chen commented Apr 27, 2024

mas-chen left a comment

boring-cyborg bot commented Apr 29, 2024

[FLINK-35228][Connectors/Kafka] Fix: DynamicKafkaSource does not read re-added topic for the same cluster #97

[FLINK-35228][Connectors/Kafka] Fix: DynamicKafkaSource does not read re-added topic for the same cluster #97

Conversation

IgnasD commented Apr 24, 2024

boring-cyborg bot commented Apr 24, 2024

IgnasD commented Apr 24, 2024

mas-chen left a comment

Choose a reason for hiding this comment

IgnasD commented Apr 25, 2024

mas-chen commented Apr 26, 2024

mas-chen left a comment

Choose a reason for hiding this comment

IgnasD commented Apr 26, 2024

mas-chen commented Apr 27, 2024

mas-chen left a comment

Choose a reason for hiding this comment

boring-cyborg bot commented Apr 29, 2024