You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2023/11/27 21:30:47 UTC
(kafka) branch trunk updated: MINOR: Rename method sendBrokerHeartbeat #14658
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 937578be657 MINOR: Rename method sendBrokerHeartbeat #14658
937578be657 is described below
commit 937578be657f309beb0a9dfe40dce5372fd35408
Author: yuyli <11...@users.noreply.github.com>
AuthorDate: Mon Nov 27 13:30:40 2023 -0800
MINOR: Rename method sendBrokerHeartbeat #14658
Rename sendBrokerHeartbeat to sendBrokerHeartbeatToUnfenceBrokers.
Reviewers: Colin P. McCabe <cm...@apache.org>
---
.../QuorumControllerIntegrationTestUtils.java | 2 +-
.../kafka/controller/QuorumControllerTest.java | 20 ++++++++++----------
2 files changed, 11 insertions(+), 11 deletions(-)
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java
index 13ac8aa9ca2..ed8daec6baf 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java
@@ -127,7 +127,7 @@ public class QuorumControllerIntegrationTestUtils {
* @param brokers The broker IDs to send heartbeats for.
* @param brokerEpochs A map from broker ID to broker epoch.
*/
- static void sendBrokerHeartbeat(
+ static void sendBrokerHeartbeatToUnfenceBrokers(
QuorumController controller,
List<Integer> brokers,
Map<Integer, Long> brokerEpochs
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
index 0fcee056684..b1145b88981 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
@@ -158,7 +158,7 @@ import static org.apache.kafka.controller.QuorumControllerIntegrationTestUtils.b
import static org.apache.kafka.controller.QuorumControllerIntegrationTestUtils.forceRenounce;
import static org.apache.kafka.controller.QuorumControllerIntegrationTestUtils.pause;
import static org.apache.kafka.controller.QuorumControllerIntegrationTestUtils.registerBrokersAndUnfence;
-import static org.apache.kafka.controller.QuorumControllerIntegrationTestUtils.sendBrokerHeartbeat;
+import static org.apache.kafka.controller.QuorumControllerIntegrationTestUtils.sendBrokerHeartbeatToUnfenceBrokers;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -304,7 +304,7 @@ public class QuorumControllerTest {
});
// Unfence all brokers and create a topic foo
- sendBrokerHeartbeat(active, allBrokers, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, allBrokers, brokerEpochs);
CreateTopicsRequestData createTopicsRequestData = new CreateTopicsRequestData().setTopics(
new CreatableTopicCollection(Collections.singleton(
new CreatableTopic().setName("foo").setNumPartitions(numberOfPartitions).
@@ -317,7 +317,7 @@ public class QuorumControllerTest {
// Fence some of the brokers
TestUtils.waitForCondition(() -> {
- sendBrokerHeartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, brokersToKeepUnfenced, brokerEpochs);
for (Integer brokerId : brokersToFence) {
if (active.clusterControl().isUnfenced(brokerId)) {
return false;
@@ -329,7 +329,7 @@ public class QuorumControllerTest {
);
// Send another heartbeat to the brokers we want to keep alive
- sendBrokerHeartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, brokersToKeepUnfenced, brokerEpochs);
// At this point only the brokers we want fenced should be fenced.
brokersToKeepUnfenced.forEach(brokerId -> {
@@ -404,7 +404,7 @@ public class QuorumControllerTest {
});
// Unfence all brokers and create a topic foo
- sendBrokerHeartbeat(active, allBrokers, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, allBrokers, brokerEpochs);
CreateTopicsRequestData createTopicsRequestData = new CreateTopicsRequestData().setTopics(
new CreatableTopicCollection(Collections.singleton(
new CreatableTopic().setName("foo").setNumPartitions(numberOfPartitions).
@@ -417,7 +417,7 @@ public class QuorumControllerTest {
// Fence some of the brokers
TestUtils.waitForCondition(
() -> {
- sendBrokerHeartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, brokersToKeepUnfenced, brokerEpochs);
for (Integer brokerId : brokersToFence) {
if (active.clusterControl().isUnfenced(brokerId)) {
return false;
@@ -430,7 +430,7 @@ public class QuorumControllerTest {
);
// Send another heartbeat to the brokers we want to keep alive
- sendBrokerHeartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, brokersToKeepUnfenced, brokerEpochs);
// At this point only the brokers we want fenced should be fenced.
brokersToKeepUnfenced.forEach(brokerId -> {
@@ -459,7 +459,7 @@ public class QuorumControllerTest {
}
// Unfence all brokers
- sendBrokerHeartbeat(active, allBrokers, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, allBrokers, brokerEpochs);
// Let the unfenced broker, 3, join the ISR partition 2
Set<TopicIdPartition> imbalancedPartitions = active.replicationControl().imbalancedPartitions();
@@ -485,14 +485,14 @@ public class QuorumControllerTest {
.Builder(alterPartitionRequest, false).build((short) 0).data()).get();
AtomicLong lastHeartbeatMs = new AtomicLong(getMonotonicMs(active.time()));
- sendBrokerHeartbeat(active, allBrokers, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, allBrokers, brokerEpochs);
// Check that partitions are balanced
TestUtils.waitForCondition(
() -> {
long currentMonotonicMs = getMonotonicMs(active.time());
if (currentMonotonicMs > lastHeartbeatMs.get() + (sessionTimeoutMillis / 2)) {
lastHeartbeatMs.set(currentMonotonicMs);
- sendBrokerHeartbeat(active, allBrokers, brokerEpochs);
+ sendBrokerHeartbeatToUnfenceBrokers(active, allBrokers, brokerEpochs);
}
return !active.replicationControl().arePartitionLeadersImbalanced();
},