Skip to content

Commit c2ca1d7

Browse files
georgew5656georgew5656
committed
[cruise-control] add flag to stop setting broker throttles
1 parent 66e625b commit c2ca1d7

4 files changed

Lines changed: 71 additions & 49 deletions

File tree

‎cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/ExecutorConfig.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,15 @@ public final class ExecutorConfig {
9393
public static final String DEFAULT_REPLICATION_THROTTLE_DOC = "The replication throttle applied to replicas being "
9494
+ "moved, in bytes per second.";
9595

96+
/**
97+
* <code>skip.replication.throttle.rate.setting</code>
98+
*/
99+
public static final String SKIP_REPLICATION_THROTTLE_RATE_SETTING_CONFIG = "skip.replication.throttle.rate.setting";
100+
public static final boolean DEFAULT_SKIP_REPLICATION_THROTTLE_RATE_SETTING = false;
101+
public static final String SKIP_REPLICATION_THROTTLE_RATE_SETTING_DOC = "If true, Cruise Control will set throttled "
102+
+ "replicas on topics during rebalances but will not set or remove throttle rates on brokers. This allows an "
103+
+ "external system to manage per-broker throttle rates independently.";
104+
96105
/**
97106
* <code>replica.movement.strategies</code>
98107
*/
@@ -548,6 +557,11 @@ public static ConfigDef define(ConfigDef configDef) {
548557
DEFAULT_DEFAULT_REPLICATION_THROTTLE,
549558
ConfigDef.Importance.MEDIUM,
550559
DEFAULT_REPLICATION_THROTTLE_DOC)
560+
.define(SKIP_REPLICATION_THROTTLE_RATE_SETTING_CONFIG,
561+
ConfigDef.Type.BOOLEAN,
562+
DEFAULT_SKIP_REPLICATION_THROTTLE_RATE_SETTING,
563+
ConfigDef.Importance.MEDIUM,
564+
SKIP_REPLICATION_THROTTLE_RATE_SETTING_DOC)
551565
.define(REPLICA_MOVEMENT_STRATEGIES_CONFIG,
552566
ConfigDef.Type.LIST,
553567
DEFAULT_REPLICA_MOVEMENT_STRATEGIES,

‎cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/Executor.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1606,8 +1606,9 @@ private void updateOngoingExecutionState() {
16061606

16071607
private void interBrokerMoveReplicas() throws InterruptedException, ExecutionException, TimeoutException {
16081608
Set<Integer> currentDeadBrokersWithReplicas = _loadMonitor.deadBrokersWithReplicas(MAX_METADATA_WAIT_MS);
1609+
boolean skipThrottleRateSetting = _config.getBoolean(ExecutorConfig.SKIP_REPLICATION_THROTTLE_RATE_SETTING_CONFIG);
16091610
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, _replicationThrottle,
1610-
currentDeadBrokersWithReplicas);
1611+
skipThrottleRateSetting, currentDeadBrokersWithReplicas);
16111612
int numTotalPartitionMovements = _executionTaskManager.numRemainingInterBrokerPartitionMovements();
16121613
long totalDataToMoveInMB = _executionTaskManager.remainingInterBrokerDataToMoveInMB();
16131614
long startTime = System.currentTimeMillis();

‎cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ReplicationThrottleHelper.java‎

Lines changed: 47 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -48,43 +48,48 @@ class ReplicationThrottleHelper {
4848
private static final int DEFAULT_RETRY_BACKOFF_BASE = 2;
4949
private final AdminClient _adminClient;
5050
private final Long _throttleRate;
51+
private final boolean _skipThrottleRateSetting;
5152
private final int _retries;
5253
private long _maxDelayMs;
5354
private final Set<Integer> _deadBrokers;
5455

55-
ReplicationThrottleHelper(AdminClient adminClient, Long throttleRate) {
56-
this(adminClient, throttleRate, RETRIES, MAX_DELAY_MS);
56+
ReplicationThrottleHelper(AdminClient adminClient, Long throttleRate, boolean skipThrottleRateSetting) {
57+
this(adminClient, throttleRate, skipThrottleRateSetting, RETRIES, MAX_DELAY_MS, new HashSet<>());
5758
}
5859

59-
ReplicationThrottleHelper(AdminClient adminClient, Long throttleRate, Set<Integer> deadBrokers) {
60-
this(adminClient, throttleRate, RETRIES, MAX_DELAY_MS, deadBrokers);
60+
ReplicationThrottleHelper(AdminClient adminClient, Long throttleRate, boolean skipThrottleRateSetting,
61+
Set<Integer> deadBrokers) {
62+
this(adminClient, throttleRate, skipThrottleRateSetting, RETRIES, MAX_DELAY_MS, deadBrokers);
6163
}
6264

6365
// for testing
64-
ReplicationThrottleHelper(AdminClient adminClient, Long throttleRate, int retries, long maxDelayMs) {
65-
this._adminClient = adminClient;
66-
this._throttleRate = throttleRate;
67-
this._retries = retries;
68-
this._maxDelayMs = maxDelayMs;
69-
this._deadBrokers = new HashSet<Integer>();
66+
ReplicationThrottleHelper(AdminClient adminClient, Long throttleRate, boolean skipThrottleRateSetting,
67+
int retries, long maxDelayMs) {
68+
this(adminClient, throttleRate, skipThrottleRateSetting, retries, maxDelayMs, new HashSet<>());
7069
}
7170

72-
ReplicationThrottleHelper(AdminClient adminClient, Long throttleRate, int retries, long maxDelayMs, Set<Integer> deadBrokers) {
71+
ReplicationThrottleHelper(AdminClient adminClient, Long throttleRate, boolean skipThrottleRateSetting,
72+
int retries, long maxDelayMs, Set<Integer> deadBrokers) {
7373
this._adminClient = adminClient;
7474
this._throttleRate = throttleRate;
75+
this._skipThrottleRateSetting = skipThrottleRateSetting;
7576
this._retries = retries;
7677
this._maxDelayMs = maxDelayMs;
7778
this._deadBrokers = deadBrokers;
7879
}
7980

8081
void setThrottles(List<ExecutionProposal> replicaMovementProposals)
8182
throws ExecutionException, InterruptedException, TimeoutException {
82-
if (throttlingEnabled()) {
83-
LOG.info("Setting a rebalance throttle of {} bytes/sec", _throttleRate);
84-
Set<Integer> participatingBrokers = getParticipatingBrokers(replicaMovementProposals);
83+
if (throttlingEnabled() || _skipThrottleRateSetting) {
8584
Map<String, Set<String>> throttledReplicas = getThrottledReplicasByTopic(replicaMovementProposals);
86-
for (int broker : participatingBrokers) {
87-
setThrottledRateIfNecessary(broker);
85+
if (throttlingEnabled() && !_skipThrottleRateSetting) {
86+
LOG.info("Setting a rebalance throttle of {} bytes/sec", _throttleRate);
87+
Set<Integer> participatingBrokers = getParticipatingBrokers(replicaMovementProposals);
88+
for (int broker : participatingBrokers) {
89+
setThrottledRateIfNecessary(broker);
90+
}
91+
} else {
92+
LOG.info("Skipping throttle rate setting; throttled replicas will still be set");
8893
}
8994
for (Map.Entry<String, Set<String>> entry : throttledReplicas.entrySet()) {
9095
setThrottledReplicas(entry.getKey(), entry.getValue());
@@ -112,7 +117,7 @@ boolean taskIsInProgress(ExecutionTask task) {
112117
// clear throttles for a specific list of execution tasks
113118
void clearThrottles(List<ExecutionTask> completedTasks, List<ExecutionTask> inProgressTasks)
114119
throws ExecutionException, InterruptedException, TimeoutException {
115-
if (throttlingEnabled()) {
120+
if (throttlingEnabled() || _skipThrottleRateSetting) {
116121
List<ExecutionProposal> completedProposals =
117122
completedTasks
118123
.stream()
@@ -121,29 +126,31 @@ void clearThrottles(List<ExecutionTask> completedTasks, List<ExecutionTask> inPr
121126
.map(ExecutionTask::proposal)
122127
.collect(Collectors.toList());
123128

124-
// These are the brokers which have completed a task with
125-
// inter-broker replica movement
126-
Set<Integer> participatingBrokers = getParticipatingBrokers(completedProposals);
127-
128-
List<ExecutionProposal> inProgressProposals =
129-
inProgressTasks
130-
.stream()
131-
.filter(this::taskIsInProgress)
132-
.map(ExecutionTask::proposal)
133-
.collect(Collectors.toList());
134-
135-
// These are the brokers which currently have in-progress
136-
// inter-broker replica movement
137-
Set<Integer> brokersWithInProgressTasks = getParticipatingBrokers(inProgressProposals);
138-
139-
// Remove the brokers with in-progress replica moves from the brokers that have
140-
// completed inter-broker replica moves
141-
Set<Integer> brokersToRemoveThrottlesFrom = new TreeSet<>(participatingBrokers);
142-
brokersToRemoveThrottlesFrom.removeAll(brokersWithInProgressTasks);
143-
144-
LOG.info("Removing replica movement throttles from brokers in the cluster: {}", brokersToRemoveThrottlesFrom);
145-
for (int broker : brokersToRemoveThrottlesFrom) {
146-
removeThrottledRateFromBroker(broker);
129+
if (throttlingEnabled() && !_skipThrottleRateSetting) {
130+
// These are the brokers which have completed a task with
131+
// inter-broker replica movement
132+
Set<Integer> participatingBrokers = getParticipatingBrokers(completedProposals);
133+
134+
List<ExecutionProposal> inProgressProposals =
135+
inProgressTasks
136+
.stream()
137+
.filter(this::taskIsInProgress)
138+
.map(ExecutionTask::proposal)
139+
.collect(Collectors.toList());
140+
141+
// These are the brokers which currently have in-progress
142+
// inter-broker replica movement
143+
Set<Integer> brokersWithInProgressTasks = getParticipatingBrokers(inProgressProposals);
144+
145+
// Remove the brokers with in-progress replica moves from the brokers that have
146+
// completed inter-broker replica moves
147+
Set<Integer> brokersToRemoveThrottlesFrom = new TreeSet<>(participatingBrokers);
148+
brokersToRemoveThrottlesFrom.removeAll(brokersWithInProgressTasks);
149+
150+
LOG.info("Removing replica movement throttles from brokers in the cluster: {}", brokersToRemoveThrottlesFrom);
151+
for (int broker : brokersToRemoveThrottlesFrom) {
152+
removeThrottledRateFromBroker(broker);
153+
}
147154
}
148155

149156
Map<String, Set<String>> throttledReplicas = getThrottledReplicasByTopic(completedProposals);

‎cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/executor/ReplicationThrottleHelperTest.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,7 @@ public void testIsNoOpWhenThrottleIsNull() throws Exception {
116116
EasyMock.replay(mockAdminClient);
117117

118118
// Test would fail on any unexpected interactions with the adminClient
119-
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, null);
119+
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, null, false);
120120
ExecutionProposal proposal = new ExecutionProposal(new TopicPartition("topic", 0),
121121
100,
122122
new ReplicaPlacementInfo(0),
@@ -146,7 +146,7 @@ public void testClearThrottleOnNonExistentTopic() throws Exception {
146146
Arrays.asList(new ReplicaPlacementInfo(brokerId0), new ReplicaPlacementInfo(brokerId2)));
147147

148148
AdminClient mockAdminClient = EasyMock.mock(AdminClient.class);
149-
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, throttleRate);
149+
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, throttleRate, false);
150150

151151
// Case 1: a situation where Topic0 does not exist. Hence no property is returned upon read.
152152

@@ -212,7 +212,7 @@ public void testSetThrottleOnNonExistentTopic() throws Exception {
212212
Arrays.asList(new ReplicaPlacementInfo(brokerId0), new ReplicaPlacementInfo(brokerId2)));
213213

214214
AdminClient mockAdminClient = EasyMock.strictMock(AdminClient.class);
215-
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, throttleRate);
215+
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, throttleRate, false);
216216

217217
// Case 1: a situation where Topic0 does not exist. Hence no property is returned upon read.
218218
expectDescribeBrokerConfigs(mockAdminClient, brokers);
@@ -246,7 +246,7 @@ public void testAddingThrottlesWithNoPreExistingThrottles() throws Exception {
246246
createTopics();
247247

248248
final long throttleRate = 100L;
249-
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, throttleRate);
249+
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, throttleRate, false);
250250
ExecutionProposal proposal = new ExecutionProposal(new TopicPartition(TOPIC0, 0),
251251
100,
252252
new ReplicaPlacementInfo(0),
@@ -278,7 +278,7 @@ public void testAddingThrottlesWithPreExistingThrottles() throws Exception {
278278
createTopics();
279279

280280
final long throttleRate = 100L;
281-
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, throttleRate);
281+
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, throttleRate, false);
282282
ExecutionProposal proposal = new ExecutionProposal(
283283
new TopicPartition(TOPIC0, 0),
284284
100,
@@ -352,7 +352,7 @@ public void testDoNotModifyExistingWildcardReplicaThrottles() throws Exception {
352352

353353
final long throttleRate = 100L;
354354

355-
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, throttleRate);
355+
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, throttleRate, false);
356356

357357
// Set replica throttle config values for both topics
358358
setWildcardThrottleReplicaForTopic(throttleHelper, TOPIC0);
@@ -423,7 +423,7 @@ public void testDoNotRemoveThrottlesForInProgressTasks() throws Exception {
423423

424424
final long throttleRate = 100L;
425425

426-
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, throttleRate);
426+
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(_adminClient, throttleRate, false);
427427
ExecutionProposal proposal = new ExecutionProposal(new TopicPartition(TOPIC0, 0),
428428
100,
429429
new ReplicaPlacementInfo(0),
@@ -497,7 +497,7 @@ public void testWaitForConfigs() throws Exception {
497497
expectDescribeTopicConfigs(mockAdminClient, TOPIC0, EMPTY_CONFIG, true);
498498
}
499499
EasyMock.replay(mockAdminClient);
500-
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, 100L, maxDelayMs, retries);
500+
ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, 100L, false, retries, maxDelayMs);
501501
ConfigResource cf = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC0);
502502
assertThrows(IllegalStateException.class, () -> throttleHelper.waitForConfigs(cf, Collections.singletonList(
503503
new AlterConfigOp(new ConfigEntry("k", "v"), AlterConfigOp.OpType.SET)

0 commit comments

Comments
 (0)