You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "RaidenE1 (via GitHub)" <gi...@apache.org> on 2025/09/25 13:35:43 UTC
[PR] MINOR:remove stalling instance [kafka]
RaidenE1 opened a new pull request, #20592:
URL: https://github.com/apache/kafka/pull/20592
Remove stalling instance in EOSIntegrationTest, since it doesn’t matter what it thinks what the assignment is but blocks the test with streams group protocol
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR:remove stalling instance [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #20592:
URL: https://github.com/apache/kafka/pull/20592#discussion_r2382260120
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -711,9 +726,8 @@ public void shouldNotViolateEosIfOneTaskGetsFencedUsingIsolatedAppInstances(fina
expectedCommittedRecordsAfterRebalance,
"The all committed records after rebalance do not match what expected");
- LOG.info("Releasing Stall");
doStall = false;
- // Once the stalling host rejoins the group, we expect both instances to see both instances.
+ // After removing the stalling instance, we expect both remaining instances to see both instances.
// It doesn't really matter what the assignment is, but we might as well also assert that they
// both see both partitions assigned exactly once
Review Comment:
The comment mentions 'both remaining instances' but the code only has one remaining instance after the stalling instance is removed. This is inconsistent with the actual implementation where only one instance remains active.
```suggestion
// After removing the stalling instance, we expect the single remaining instance to see only itself as a member.
// It doesn't really matter what the assignment is, but we might as well also assert that it
// sees both partitions assigned exactly once
```
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -605,12 +626,11 @@ public void shouldNotViolateEosIfOneTaskGetsFencedUsingIsolatedAppInstances(fina
// -> the stall only affects one thread and should trigger a rebalance
// after rebalancing, we should read 40 committed records (even if 50 record got written)
//
- // afterward, the "stalling" thread resumes, and another rebalance should get triggered
- // we write the remaining 20 records and verify to read 60 result records
+ // after the stalling instance is removed, we write the remaining 20 records and verify to read 60 result records
Review Comment:
The comment mentions 'both remaining instances' but the code only has one remaining instance after the stalling instance is removed. This is inconsistent with the actual implementation where only one instance remains active.
```suggestion
// after the stalling instance is removed, the remaining instance writes the remaining 20 records and we verify to read 60 result records
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR:remove stalling instance [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20592:
URL: https://github.com/apache/kafka/pull/20592#discussion_r2382259714
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -711,9 +726,8 @@ public void shouldNotViolateEosIfOneTaskGetsFencedUsingIsolatedAppInstances(fina
expectedCommittedRecordsAfterRebalance,
"The all committed records after rebalance do not match what expected");
- LOG.info("Releasing Stall");
Review Comment:
why have you removed this
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -711,9 +726,8 @@ public void shouldNotViolateEosIfOneTaskGetsFencedUsingIsolatedAppInstances(fina
expectedCommittedRecordsAfterRebalance,
"The all committed records after rebalance do not match what expected");
- LOG.info("Releasing Stall");
doStall = false;
- // Once the stalling host rejoins the group, we expect both instances to see both instances.
+ // After removing the stalling instance, we expect both remaining instances to see both instances.
Review Comment:
why are you changin this comment
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -181,16 +190,19 @@ public void createTopics() throws Exception {
CLUSTER.createTopic(MULTI_PARTITION_INPUT_TOPIC, NUM_TOPIC_PARTITIONS, 1);
CLUSTER.createTopic(MULTI_PARTITION_THROUGH_TOPIC, NUM_TOPIC_PARTITIONS, 1);
CLUSTER.createTopic(MULTI_PARTITION_OUTPUT_TOPIC, NUM_TOPIC_PARTITIONS, 1);
+ CLUSTER.setGroupStandbyReplicas(applicationId, 1);
Review Comment:
INteresting that this works even if we use the old protocol.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -667,24 +687,19 @@ public void shouldNotViolateEosIfOneTaskGetsFencedUsingIsolatedAppInstances(fina
"Expected a host to start stalling"
);
final String observedStallingHost = stallingHost.get();
- final KafkaStreams stallingInstance;
final KafkaStreams remainingInstance;
if ("streams1".equals(observedStallingHost)) {
- stallingInstance = streams1;
remainingInstance = streams2;
} else if ("streams2".equals(observedStallingHost)) {
- stallingInstance = streams2;
remainingInstance = streams1;
} else {
throw new IllegalArgumentException("unexpected host name: " + observedStallingHost);
}
- // the stalling instance won't have an updated view, and it doesn't matter what it thinks
- // the assignment is. We only really care that the remaining instance only sees one host
- // that owns both partitions.
+ // After removing the stalling instance, we only care that the remaining instance
+ // sees one host that owns both partitions.
waitForCondition(
- () -> stallingInstance.metadataForAllStreamsClients().size() == 2
- && remainingInstance.metadataForAllStreamsClients().size() == 1
+ () -> remainingInstance.metadataForAllStreamsClients().size() == 1
Review Comment:
cc @bbejeck
We have a difference between IQv2 in old protocol and new protocol in this test.
We drop a "stalling instance" from the group. Then we check that the remaining instance has correctly reflected the change (all partitions are served by the remaining instance).
What the stalling instance has in it's IQv2 metadata doesn't really matter, since it dropped from the group. There is also a comment here:
```
// the stalling instance won't have an updated view, and it doesn't matter what it thinks
// the assignment is.
```
However, we still seem to check it's view:
```
stallingInstance.metadataForAllStreamsClients().size() == 2
```
There is a difference in timing here, I think the stalling instance never sees the second instance (maybe down to hearbeat interval).
So my thinking is that we can remove this check
```
stallingInstance.metadataForAllStreamsClients().size() == 2
```
Just asking to validate my thinking.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java:
##########
@@ -667,24 +687,19 @@ public void shouldNotViolateEosIfOneTaskGetsFencedUsingIsolatedAppInstances(fina
"Expected a host to start stalling"
);
final String observedStallingHost = stallingHost.get();
- final KafkaStreams stallingInstance;
final KafkaStreams remainingInstance;
if ("streams1".equals(observedStallingHost)) {
- stallingInstance = streams1;
remainingInstance = streams2;
} else if ("streams2".equals(observedStallingHost)) {
- stallingInstance = streams2;
remainingInstance = streams1;
} else {
throw new IllegalArgumentException("unexpected host name: " + observedStallingHost);
}
- // the stalling instance won't have an updated view, and it doesn't matter what it thinks
- // the assignment is. We only really care that the remaining instance only sees one host
- // that owns both partitions.
+ // After removing the stalling instance, we only care that the remaining instance
Review Comment:
@RaidenE1 Why have you removed the comment about the stalling instance
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Enable streams rebalance protocol in EosIntegrationTest [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #20592:
URL: https://github.com/apache/kafka/pull/20592
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org