[FLINK-26645][Connector/Pulsar] Fix Pulsar source subscriber consume from all partitions when only subscribed to 1 partition #19092
+139
−59
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
imaffe
changed the title
imaffe
commented
Mar 15, 2022
| String name = topicName.getPartitionedTopicName(); | ||
| int index = topicName.getPartitionIndex(); | ||
|
|
||
| TopicMetadata metadata = queryTopicMetadata(pulsarAdmin, name); |
There was a problem hiding this comment.
For a partition, should the topicMetadata be created manually or query from the admin api ?
There was a problem hiding this comment.
Query from admin API is ok.
|
BK-1.14, BK-1.15 are required. |
syhily
reviewed
Mar 15, 2022
| } | ||
|
|
||
| public static boolean isPartitioned(String topic) { | ||
| public static boolean isPartitionOfPartitionedTopic(String topic) { |
There was a problem hiding this comment.
Sure, this is a more concise naming~
syhily
force-pushed
the
FLINK-26645
branch
4 times, most recently
from
ffc7781 to
0cffb7e
Compare
AHeise
reviewed
Apr 1, 2022
Comment on lines
71
to
72
| operator().deleteTopic(topic1); | ||
| operator().deleteTopic(topic2); |
There was a problem hiding this comment.
these should be executed in a finally statement or else a failing test will leave the topic.
Alternatively, if topic creation is not taking too long, just create and delete all topics in @beforeeach and @AfterEach. Do we even need to create them for each test? Maybe BeforeAll is sufficient?
There was a problem hiding this comment.
I think you are right, I have fixed this issue.
AHeise
approved these changes
Apr 3, 2022
This was referenced Apr 12, 2022
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
Fix FLINK-26645: When users specify only consume from 1 partition but the source consumes from all partitions.
Brief change log
Verifying this change
This change added tests and can be verified as follows:
Does this pull request potentially affect one of the following parts:
No
Documentation
No