You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ahuang98 (via GitHub)" <gi...@apache.org> on 2024/09/10 01:30:44 UTC
[PR] KAFKA-16963 followup: Ducktape test for KIP-853 [kafka]
ahuang98 opened a new pull request, #17146:
URL: https://github.com/apache/kafka/pull/17146
Addressing Jose' comments in https://github.com/apache/kafka/pull/17081
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17146:
URL: https://github.com/apache/kafka/pull/17146#discussion_r1776104122
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1065,6 +1065,11 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
int epoch = batch.epoch();
List<ApiMessageAndVersion> messages = batch.records();
+ if (messages.isEmpty()) {
+ log.debug("Skipping handling commit for batch with no data records with offset {} and epoch {}.", offset, epoch);
+ continue;
Review Comment:
Also, we should update the metrics in this case (should be easy to do by adding a new method to OffsetsControlManager to do just that and nothing else)
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #17146:
URL: https://github.com/apache/kafka/pull/17146#issuecomment-2377607594
> It looks like upgrade tests are complaining about advertised listeners including controller listeners (when broker is in the process role). I'll adjust the test framework to exclude when it's not a dynamicRaftQuorum for now
If the process role is broker only then the controller's listener should not be in the advertise listener.
--
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] KAFKA-16963 followup: Ducktape test for KIP-853 [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #17146:
URL: https://github.com/apache/kafka/pull/17146#discussion_r1757400655
##########
tests/kafkatest/tests/core/quorum_reconfiguration_test.py:
##########
@@ -46,54 +50,67 @@ def setUp(self):
self.num_producers = 1
self.num_consumers = 1
- def perform_reconfig(self, active_controller_id, inactive_controller_id, inactive_controller, broker_ids):
+ def perform_reconfig(self,
+ active_controller_id: int,
+ inactive_controller_id: int,
+ inactive_controller: ClusterNode,
+ broker_only_ids: List[int]):
+ """
+ Tests quorum reconfiguration by adding a second controller and then removing the active controller.
+
+ :param active_controller_id: id of the active controller
+ :param inactive_controller_id: id of the inactive controller
+ :param inactive_controller: node object of the inactive controller
+ :param broker_only_ids: broker ids of nodes which have no controller process
+ """
# Check describe quorum output shows the controller (first node) is the leader and the only voter
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id],
+ broker_only_ids), timeout_sec=5)
# Start second controller
self.kafka.controller_quorum.add_broker(inactive_controller)
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids + [inactive_controller_id])
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id],
+ broker_only_ids + [inactive_controller_id]), timeout_sec=5)
# Add controller to quorum
self.kafka.controller_quorum.add_controller(inactive_controller_id, inactive_controller)
# Check describe quorum output shows both controllers are voters
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id, inactive_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id, inactive_controller_id],
+ broker_only_ids), timeout_sec=5)
# Remove leader from quorum
- voters = json_from_line(r"CurrentVoters:.*", output)
+ voters = json_from_line(r"CurrentVoters:.*", self.kafka.describe_quorum())
directory_id = next(voter["directoryId"] for voter in voters if voter["id"] == active_controller_id)
self.kafka.controller_quorum.remove_controller(active_controller_id, directory_id)
-
- # Check describe quorum output to show second_controller is now the leader
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(inactive_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, inactive_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ # Check describe quorum output to show second_controller is now the leader. Observers contain the old controller
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ inactive_controller_id,
+ [inactive_controller_id],
+ broker_only_ids), timeout_sec=5)
Review Comment:
The reason why the leader is not in the observer is related to my other comment. `check_describe_quorum_output` always returns true.
##########
tests/kafkatest/tests/core/quorum_reconfiguration_test.py:
##########
@@ -147,17 +170,31 @@ def test_isolated_mode_reconfig(self, metadata_quorum):
inactive_controller,
[self.kafka.idx(node) for node in self.kafka.nodes]))
-def assert_nodes_in_output(pattern, output, *node_ids):
+def assert_nodes_in_output(pattern: str, output: str, *node_ids: int):
nodes = json_from_line(pattern, output)
assert len(nodes) == len(node_ids)
for node in nodes:
assert node["id"] in node_ids
-def json_from_line(pattern, output):
+def check_describe_quorum_output(output: str, leader_id: int, voter_ids: List[int], observer_ids: List[int]):
+ """
+ Check that the describe quorum output contains the expected leader, voters, and observers
+ :param output: Describe quorum output
+ :param leader_id: Expected leader id
+ :param voter_ids: Expected voter ids
+ :param observer_ids: Expected observer ids
+ :return:
+ """
+ assert re.search(r"LeaderId:\s*" + str(leader_id), output)
+ assert_nodes_in_output(r"CurrentVoters:.*", output, *voter_ids)
+ assert_nodes_in_output(r"CurrentObservers:.*", output, *observer_ids)
Review Comment:
Do you want to change `assert_nodes_in_output` to `check_nodes_in_output` which instead of `assert` returns a boolean. With that you can can change the `return` statement to:
```python
return check_nodes_in_output(r"CurrentVoters:.*", output, *voter_ids) &&
check_nodes_in_output(r"CurrentObservers:.*", output, *observer_ids)
```
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on PR #17146:
URL: https://github.com/apache/kafka/pull/17146#issuecomment-2378009931
my local machine seems to be too slow to verify the fix for upgrade tests - trying to trigger a run on Semaphore off of https://github.com/apache/kafka/pull/17294
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on PR #17146:
URL: https://github.com/apache/kafka/pull/17146#issuecomment-2377576008
It looks like upgrade tests are complaining about advertised listeners including controller listeners (when broker is in the process role). I'll adjust the test framework to exclude when it's not a dynamicRaftQuorum for now
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #17146:
URL: https://github.com/apache/kafka/pull/17146#discussion_r1776073349
##########
tests/kafkatest/tests/core/quorum_reconfiguration_test.py:
##########
@@ -46,54 +50,67 @@ def setUp(self):
self.num_producers = 1
self.num_consumers = 1
- def perform_reconfig(self, active_controller_id, inactive_controller_id, inactive_controller, broker_ids):
+ def perform_reconfig(self,
+ active_controller_id: int,
+ inactive_controller_id: int,
+ inactive_controller: ClusterNode,
+ broker_only_ids: List[int]):
+ """
+ Tests quorum reconfiguration by adding a second controller and then removing the active controller.
+
+ :param active_controller_id: id of the active controller
+ :param inactive_controller_id: id of the inactive controller
+ :param inactive_controller: node object of the inactive controller
+ :param broker_only_ids: broker ids of nodes which have no controller process
+ """
# Check describe quorum output shows the controller (first node) is the leader and the only voter
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id],
+ broker_only_ids), timeout_sec=5)
# Start second controller
self.kafka.controller_quorum.add_broker(inactive_controller)
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids + [inactive_controller_id])
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id],
+ broker_only_ids + [inactive_controller_id]), timeout_sec=5)
# Add controller to quorum
self.kafka.controller_quorum.add_controller(inactive_controller_id, inactive_controller)
# Check describe quorum output shows both controllers are voters
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id, inactive_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id, inactive_controller_id],
+ broker_only_ids), timeout_sec=5)
# Remove leader from quorum
- voters = json_from_line(r"CurrentVoters:.*", output)
+ voters = json_from_line(r"CurrentVoters:.*", self.kafka.describe_quorum())
directory_id = next(voter["directoryId"] for voter in voters if voter["id"] == active_controller_id)
self.kafka.controller_quorum.remove_controller(active_controller_id, directory_id)
-
- # Check describe quorum output to show second_controller is now the leader
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(inactive_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, inactive_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ # Check describe quorum output to show second_controller is now the leader. Observers contain the old controller
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ inactive_controller_id,
+ [inactive_controller_id],
+ broker_only_ids), timeout_sec=5)
Review Comment:
we discussed offline that technically wait_until should work on asserts. we found the reason the old leader was not in the observers is due to KAFKA-17608
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17146:
URL: https://github.com/apache/kafka/pull/17146#discussion_r1776103853
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1065,6 +1065,11 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
int epoch = batch.epoch();
List<ApiMessageAndVersion> messages = batch.records();
+ if (messages.isEmpty()) {
+ log.debug("Skipping handling commit for batch with no data records with offset {} and epoch {}.", offset, epoch);
+ continue;
Review Comment:
`continue` is kind of a messy construct. Can we do something like this instead?
```
if (messages.isEmpty()) {
log.debug("Skipping handling commit for batch with no data records with offset {} and epoch {}.", offset, epoch);
} else if (isActive) {
...
```
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #17146:
URL: https://github.com/apache/kafka/pull/17146#discussion_r1776104765
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1065,6 +1065,11 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
int epoch = batch.epoch();
List<ApiMessageAndVersion> messages = batch.records();
+ if (messages.isEmpty()) {
+ log.debug("Skipping handling commit for batch with no data records with offset {} and epoch {}.", offset, epoch);
+ continue;
+ }
+
Review Comment:
Let's get rid of this line which isn't needed:
```
// The active controller can delete up to the current committed offset.
snapshotRegistry.deleteSnapshotsUpTo(offsetControl.lastStableOffset());
```
since that is already done in `OffsetsControlManager.handleCommitBatch`.
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on PR #17146:
URL: https://github.com/apache/kafka/pull/17146#issuecomment-2377635801
It looked like a version incompatibility issue since the specific error is
```
The advertised.listeners config must not contain KRaft controller listeners from controller.listener.names when process.roles contains the broker role because Kafka clients that send requests via advertised listeners do not send requests to KRaft controllers -- they only send requests to KRaft brokers.\n\tat scala.Predef$.require(Predef.scala:281)\n\tat kafka.server.KafkaConfig.validateAdvertisedListenersDoesNotContainControllerListenersForKRaftBroker$1(KafkaConfig.scala:2047)
```
which is logged from `validateAdvertisedListenersDoesNotContainControllerListenersForKRaftBroker` which was removed in https://github.com/apache/kafka/pull/16892
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on PR #17146:
URL: https://github.com/apache/kafka/pull/17146#issuecomment-2377573465
```
[INFO:2024-09-25 17:02:28,487]: starting test run with session id 2024-09-25--008...
[INFO:2024-09-25 17:02:28,488]: running 1 tests...
[INFO:2024-09-25 17:02:28,488]: Triggering test 1 of 1...
[INFO:2024-09-25 17:02:28,500]: RunnerClient: Loading test {'directory': '/opt/kafka-dev/tests/kafkatest/tests/core', 'file_name': 'quorum_reconfiguration_test.py', 'cls_name': 'TestQuorumReconfiguration', 'method_name': 'test_isolated_mode_reconfig', 'injected_args': {'metadata_quorum': 'ISOLATED_KRAFT'}}
[INFO:2024-09-25 17:02:28,503]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_isolated_mode_reconfig.metadata_quorum=ISOLATED_KRAFT: on run 1/1
[INFO:2024-09-25 17:02:28,503]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_isolated_mode_reconfig.metadata_quorum=ISOLATED_KRAFT: Setting up...
[INFO:2024-09-25 17:02:28,504]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_isolated_mode_reconfig.metadata_quorum=ISOLATED_KRAFT: Running...
[INFO:2024-09-25 17:03:37,732]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_isolated_mode_reconfig.metadata_quorum=ISOLATED_KRAFT: Tearing down...
[INFO:2024-09-25 17:03:54,910]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_isolated_mode_reconfig.metadata_quorum=ISOLATED_KRAFT: PASS
[INFO:2024-09-25 17:03:54,917]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_isolated_mode_reconfig.metadata_quorum=ISOLATED_KRAFT: Data: None
================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.11.4
session_id: 2024-09-25--008
run time: 1 minute 26.528 seconds
tests run: 1
passed: 1
flaky: 0
failed: 0
ignored: 0
================================================================================
test_id: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_isolated_mode_reconfig.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 26.408 seconds
```
and
```
[INFO:2024-09-25 15:37:13,023]: starting test run with session id 2024-09-25--005...
[INFO:2024-09-25 15:37:13,024]: running 1 tests...
[INFO:2024-09-25 15:37:13,024]: Triggering test 1 of 1...
[INFO:2024-09-25 15:37:13,033]: RunnerClient: Loading test {'directory': '/opt/kafka-dev/tests/kafkatest/tests/core', 'file_name': 'quorum_reconfiguration_test.py', 'cls_name': 'TestQuorumReconfiguration', 'method_name': 'test_combined_mode_reconfig', 'injected_args': {'metadata_quorum': 'COMBINED_KRAFT'}}
[INFO:2024-09-25 15:37:13,037]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_combined_mode_reconfig.metadata_quorum=COMBINED_KRAFT: on run 1/1
[INFO:2024-09-25 15:37:13,038]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_combined_mode_reconfig.metadata_quorum=COMBINED_KRAFT: Setting up...
[INFO:2024-09-25 15:37:13,038]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_combined_mode_reconfig.metadata_quorum=COMBINED_KRAFT: Running...
[INFO:2024-09-25 15:38:18,170]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_combined_mode_reconfig.metadata_quorum=COMBINED_KRAFT: Tearing down...
[INFO:2024-09-25 15:38:29,252]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_combined_mode_reconfig.metadata_quorum=COMBINED_KRAFT: PASS
[INFO:2024-09-25 15:38:29,254]: RunnerClient: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_combined_mode_reconfig.metadata_quorum=COMBINED_KRAFT: Data: None
================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.11.4
session_id: 2024-09-25--005
run time: 1 minute 16.249 seconds
tests run: 1
passed: 1
flaky: 0
failed: 0
ignored: 0
================================================================================
test_id: kafkatest.tests.core.quorum_reconfiguration_test.TestQuorumReconfiguration.test_combined_mode_reconfig.metadata_quorum=COMBINED_KRAFT
status: PASS
run time: 1 minute 16.215 seconds
--------------------------------------------------------------------------------
```
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #17146:
URL: https://github.com/apache/kafka/pull/17146#issuecomment-2377679464
> If the process role is broker only then the controller's listener should not be in the advertise listener.
I'm bumping up against this now when `dynamicRaftQuorum=False`:
```
ducker@ducker05: " java.lang.IllegalArgumentException: requirement failed: If process.roles contains just the 'broker' role, the node id 4 must not be included in the set of voters controller.quorum.voters=Set(4)
```
If set it to `True` the system tests run OK, but I'd prefer not do that.
--
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] KAFKA-16963 followup: Ducktape test for KIP-853 [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17146:
URL: https://github.com/apache/kafka/pull/17146#issuecomment-2351720053
@ahuang98 could you please fix the conflicts?
--
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] KAFKA-16963 followup: Ducktape test for KIP-853 [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #17146:
URL: https://github.com/apache/kafka/pull/17146#discussion_r1752300760
##########
tests/kafkatest/tests/core/quorum_reconfiguration_test.py:
##########
@@ -46,54 +50,67 @@ def setUp(self):
self.num_producers = 1
self.num_consumers = 1
- def perform_reconfig(self, active_controller_id, inactive_controller_id, inactive_controller, broker_ids):
+ def perform_reconfig(self,
+ active_controller_id: int,
+ inactive_controller_id: int,
+ inactive_controller: ClusterNode,
+ broker_only_ids: List[int]):
+ """
+ Tests quorum reconfiguration by adding a second controller and then removing the active controller.
+
+ :param active_controller_id: id of the active controller
+ :param inactive_controller_id: id of the inactive controller
+ :param inactive_controller: node object of the inactive controller
+ :param broker_only_ids: broker ids of nodes which have no controller process
+ """
# Check describe quorum output shows the controller (first node) is the leader and the only voter
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id],
+ broker_only_ids), timeout_sec=5)
# Start second controller
self.kafka.controller_quorum.add_broker(inactive_controller)
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids + [inactive_controller_id])
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id],
+ broker_only_ids + [inactive_controller_id]), timeout_sec=5)
# Add controller to quorum
self.kafka.controller_quorum.add_controller(inactive_controller_id, inactive_controller)
# Check describe quorum output shows both controllers are voters
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id, inactive_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id, inactive_controller_id],
+ broker_only_ids), timeout_sec=5)
# Remove leader from quorum
- voters = json_from_line(r"CurrentVoters:.*", output)
+ voters = json_from_line(r"CurrentVoters:.*", self.kafka.describe_quorum())
directory_id = next(voter["directoryId"] for voter in voters if voter["id"] == active_controller_id)
self.kafka.controller_quorum.remove_controller(active_controller_id, directory_id)
-
- # Check describe quorum output to show second_controller is now the leader
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(inactive_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, inactive_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ # Check describe quorum output to show second_controller is now the leader. Observers contain the old controller
Review Comment:
This needs to be re-looked at
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio merged PR #17146:
URL: https://github.com/apache/kafka/pull/17146
--
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] KAFKA-17608: KRaft controller crashes when leader is removed from quorum [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #17146:
URL: https://github.com/apache/kafka/pull/17146#discussion_r1776070346
##########
tests/kafkatest/tests/core/quorum_reconfiguration_test.py:
##########
@@ -46,54 +50,67 @@ def setUp(self):
self.num_producers = 1
self.num_consumers = 1
- def perform_reconfig(self, active_controller_id, inactive_controller_id, inactive_controller, broker_ids):
+ def perform_reconfig(self,
+ active_controller_id: int,
+ inactive_controller_id: int,
+ inactive_controller: ClusterNode,
+ broker_only_ids: List[int]):
+ """
+ Tests quorum reconfiguration by adding a second controller and then removing the active controller.
+
+ :param active_controller_id: id of the active controller
+ :param inactive_controller_id: id of the inactive controller
+ :param inactive_controller: node object of the inactive controller
+ :param broker_only_ids: broker ids of nodes which have no controller process
+ """
# Check describe quorum output shows the controller (first node) is the leader and the only voter
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id],
+ broker_only_ids), timeout_sec=5)
# Start second controller
self.kafka.controller_quorum.add_broker(inactive_controller)
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids + [inactive_controller_id])
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id],
+ broker_only_ids + [inactive_controller_id]), timeout_sec=5)
# Add controller to quorum
self.kafka.controller_quorum.add_controller(inactive_controller_id, inactive_controller)
# Check describe quorum output shows both controllers are voters
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(active_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, active_controller_id, inactive_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ active_controller_id,
+ [active_controller_id, inactive_controller_id],
+ broker_only_ids), timeout_sec=5)
# Remove leader from quorum
- voters = json_from_line(r"CurrentVoters:.*", output)
+ voters = json_from_line(r"CurrentVoters:.*", self.kafka.describe_quorum())
directory_id = next(voter["directoryId"] for voter in voters if voter["id"] == active_controller_id)
self.kafka.controller_quorum.remove_controller(active_controller_id, directory_id)
-
- # Check describe quorum output to show second_controller is now the leader
- output = self.kafka.describe_quorum()
- assert re.search(r"LeaderId:\s*" + str(inactive_controller_id), output)
- assert_nodes_in_output(r"CurrentVoters:.*", output, inactive_controller_id)
- assert_nodes_in_output(r"CurrentObservers:.*", output, *broker_ids)
-
+ # Describe quorum output shows the second controller is now leader, old controller is an observer
+ wait_until(lambda: check_describe_quorum_output(self.kafka.describe_quorum(),
+ inactive_controller_id,
+ [inactive_controller_id],
+ broker_only_ids + [active_controller_id]), timeout_sec=30)
Review Comment:
timeout here can be reduced to 5, will change with the next review
--
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