showuon commented on a change in pull request #11688:
URL: https://github.com/apache/kafka/pull/11688#discussion_r801269210
##########
File path: core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
##########
@@ -1791,4 +1793,33 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertTrue(records2.count() == 1 &&
records2.records(tp).asScala.head.offset == 1,
"Expected consumer2 to consume one message from offset 1, which is the
committed offset of consumer1")
}
+ @Test
+ def testStaticConsumerDetectsNewPartitionCreatedAfterRestart(): Unit = {
Review comment:
I didn't see any consumer restart in the test. Are you trying to say
`afterNewConsumerCreated`?
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
##########
@@ -211,7 +211,15 @@ protected void onJoinComplete(int generation, String
memberId, String protocol,
}
@Override
- protected Map<String, ByteBuffer> performAssignment(String leaderId,
String protocol, List<JoinGroupResponseMember> allMemberMetadata) {
+ protected Map<String, ByteBuffer> performAssignment(String leaderId,
+ String protocol,
+
List<JoinGroupResponseMember> allMemberMetadata,
+ Boolean
skipAssignment) {
+ // Connect does not support static membership so skipping the
+ // assignment should never happen in practice.
+ if (skipAssignment)
+ return Collections.emptyMap();
Review comment:
Agree! At least we should log a warning here.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
##########
@@ -655,11 +667,8 @@ private void maybeUpdateGroupSubscription(String
assignorName,
maybeUpdateGroupSubscription(assignorName, assignments,
allSubscribedTopics);
- assignmentSnapshot = metadataSnapshot;
Review comment:
I think the reason why we need to set `assignmentSnapShot` here is
because after assignment (ex: custom assignor), it might be possible that there
are extra topics to be added, and we need to request new metadata and add that
in to `metadataSnapshot` in `maybeUpdateGroupSubscription()` method, right?
So, I think you could do like this:
```java
if (skipAssignment) {
// set the metadataSnapshot here for skipAssignment case
assignmentSnapshot = metadataSnapshot;
return Collections.emptyMap();
}
...
maybeUpdateGroupSubscription(assignorName, assignments, allSubscribedTopics);
// still keep this line here
assignmentSnapshot = metadataSnapshot;
```
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
