Skip to content

Commit 17da75e

Browse files
LiebingYuLiebing
authored and
zcoo
committed
[server] Use batch operations of ZooKeeper to optimize updateLeaderAndIsr (apache#1445)
Co-authored-by: Liebing <[email protected]>
1 parent 1c50f15 commit 17da75e

File tree

4 files changed

+152
-11
lines changed

4 files changed

+152
-11
lines changed

‎fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorEventProcessor.java‎

Lines changed: 29 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@
9696
import java.util.ArrayList;
9797
import java.util.Arrays;
9898
import java.util.Collections;
99+
import java.util.HashMap;
99100
import java.util.HashSet;
100101
import java.util.List;
101102
import java.util.Map;
@@ -878,6 +879,7 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
878879
// TODO verify leader epoch.
879880

880881
List<AdjustIsrResultForBucket> result = new ArrayList<>();
882+
Map<TableBucket, LeaderAndIsr> newLeaderAndIsrList = new HashMap<>();
881883
for (Map.Entry<TableBucket, LeaderAndIsr> entry : leaderAndIsrList.entrySet()) {
882884
TableBucket tableBucket = entry.getKey();
883885
LeaderAndIsr tryAdjustLeaderAndIsr = entry.getValue();
@@ -910,21 +912,37 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
910912
tryAdjustLeaderAndIsr.isr(),
911913
coordinatorContext.getCoordinatorEpoch(),
912914
currentLeaderAndIsr.bucketEpoch() + 1);
913-
try {
914-
zooKeeperClient.updateLeaderAndIsr(tableBucket, newLeaderAndIsr);
915-
} catch (Exception e) {
916-
LOG.error("Error when register leader and isr.", e);
917-
result.add(new AdjustIsrResultForBucket(tableBucket, ApiError.fromThrowable(e)));
915+
newLeaderAndIsrList.put(tableBucket, newLeaderAndIsr);
916+
}
917+
918+
try {
919+
zooKeeperClient.batchUpdateLeaderAndIsr(newLeaderAndIsrList);
920+
newLeaderAndIsrList.forEach(
921+
(tableBucket, newLeaderAndIsr) ->
922+
result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr)));
923+
} catch (Exception batchException) {
924+
LOG.error("Error when batch update leader and isr. Try one by one.", batchException);
925+
926+
for (Map.Entry<TableBucket, LeaderAndIsr> entry : newLeaderAndIsrList.entrySet()) {
927+
TableBucket tableBucket = entry.getKey();
928+
LeaderAndIsr newLeaderAndIsr = entry.getValue();
929+
try {
930+
zooKeeperClient.updateLeaderAndIsr(tableBucket, newLeaderAndIsr);
931+
} catch (Exception e) {
932+
LOG.error("Error when register leader and isr.", e);
933+
result.add(
934+
new AdjustIsrResultForBucket(tableBucket, ApiError.fromThrowable(e)));
935+
}
936+
// Successful return.
937+
result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr));
918938
}
939+
}
919940

920-
// update coordinator leader and isr cache.
921-
coordinatorContext.putBucketLeaderAndIsr(tableBucket, newLeaderAndIsr);
941+
// update coordinator leader and isr cache.
942+
newLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr);
922943

923-
// TODO update metadata for all alive tablet servers.
944+
// TODO update metadata for all alive tablet servers.
924945

925-
// Successful return.
926-
result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr));
927-
}
928946
return result;
929947
}
930948

‎fluss-server/src/main/java/com/alibaba/fluss/server/zk/ZooKeeperClient.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -326,6 +326,31 @@ public void updateLeaderAndIsr(TableBucket tableBucket, LeaderAndIsr leaderAndIs
326326
LOG.info("Updated {} for bucket {} in Zookeeper.", leaderAndIsr, tableBucket);
327327
}
328328

329+
public void batchUpdateLeaderAndIsr(Map<TableBucket, LeaderAndIsr> leaderAndIsrList)
330+
throws Exception {
331+
if (leaderAndIsrList.isEmpty()) {
332+
return;
333+
}
334+
335+
List<CuratorOp> ops = new ArrayList<>(leaderAndIsrList.size());
336+
for (Map.Entry<TableBucket, LeaderAndIsr> entry : leaderAndIsrList.entrySet()) {
337+
TableBucket tableBucket = entry.getKey();
338+
LeaderAndIsr leaderAndIsr = entry.getValue();
339+
340+
String path = LeaderAndIsrZNode.path(tableBucket);
341+
byte[] data = LeaderAndIsrZNode.encode(leaderAndIsr);
342+
CuratorOp updateOp = zkClient.transactionOp().setData().forPath(path, data);
343+
ops.add(updateOp);
344+
if (ops.size() == MAX_BATCH_SIZE) {
345+
zkClient.transaction().forOperations(ops);
346+
ops.clear();
347+
}
348+
}
349+
if (!ops.isEmpty()) {
350+
zkClient.transaction().forOperations(ops);
351+
}
352+
}
353+
329354
public void deleteLeaderAndIsr(TableBucket tableBucket) throws Exception {
330355
String path = LeaderAndIsrZNode.path(tableBucket);
331356
zkClient.delete().forPath(path);

‎fluss-server/src/test/java/com/alibaba/fluss/server/coordinator/CoordinatorEventProcessorTest.java‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,16 +31,19 @@
3131
import com.alibaba.fluss.metadata.TableDescriptor;
3232
import com.alibaba.fluss.metadata.TablePartition;
3333
import com.alibaba.fluss.metadata.TablePath;
34+
import com.alibaba.fluss.rpc.messages.AdjustIsrResponse;
3435
import com.alibaba.fluss.rpc.messages.CommitKvSnapshotResponse;
3536
import com.alibaba.fluss.rpc.messages.CommitRemoteLogManifestResponse;
3637
import com.alibaba.fluss.rpc.messages.NotifyKvSnapshotOffsetRequest;
3738
import com.alibaba.fluss.rpc.messages.NotifyRemoteLogOffsetsRequest;
3839
import com.alibaba.fluss.server.coordinator.event.AccessContextEvent;
40+
import com.alibaba.fluss.server.coordinator.event.AdjustIsrReceivedEvent;
3941
import com.alibaba.fluss.server.coordinator.event.CommitKvSnapshotEvent;
4042
import com.alibaba.fluss.server.coordinator.event.CommitRemoteLogManifestEvent;
4143
import com.alibaba.fluss.server.coordinator.event.CoordinatorEventManager;
4244
import com.alibaba.fluss.server.coordinator.statemachine.BucketState;
4345
import com.alibaba.fluss.server.coordinator.statemachine.ReplicaState;
46+
import com.alibaba.fluss.server.entity.AdjustIsrResultForBucket;
4447
import com.alibaba.fluss.server.entity.CommitKvSnapshotData;
4548
import com.alibaba.fluss.server.entity.CommitRemoteLogManifestData;
4649
import com.alibaba.fluss.server.kv.snapshot.CompletedSnapshot;
@@ -78,6 +81,7 @@
7881
import java.time.Duration;
7982
import java.util.Arrays;
8083
import java.util.Collections;
84+
import java.util.HashMap;
8185
import java.util.List;
8286
import java.util.Map;
8387
import java.util.Optional;
@@ -99,6 +103,7 @@
99103
import static com.alibaba.fluss.server.coordinator.statemachine.ReplicaState.OfflineReplica;
100104
import static com.alibaba.fluss.server.coordinator.statemachine.ReplicaState.OnlineReplica;
101105
import static com.alibaba.fluss.server.testutils.KvTestUtils.mockCompletedSnapshot;
106+
import static com.alibaba.fluss.server.utils.ServerRpcMessageUtils.getAdjustIsrResponseData;
102107
import static com.alibaba.fluss.server.utils.TableAssignmentUtils.generateAssignment;
103108
import static com.alibaba.fluss.testutils.common.CommonTestUtils.retry;
104109
import static com.alibaba.fluss.testutils.common.CommonTestUtils.waitValue;
@@ -761,6 +766,59 @@ void testNotifyOffsetsWithShrinkISR(@TempDir Path tempDir) throws Exception {
761766
verifyReceiveRequestExceptFor(3, leader, NotifyKvSnapshotOffsetRequest.class);
762767
}
763768

769+
@Test
770+
void testProcessAdjustIsr() throws Exception {
771+
// make sure all request to gateway should be successful
772+
initCoordinatorChannel();
773+
// create a table,
774+
TablePath t1 = TablePath.of(defaultDatabase, "create_process_adjust_isr");
775+
int nBuckets = 3;
776+
int replicationFactor = 3;
777+
TableAssignment tableAssignment =
778+
generateAssignment(
779+
nBuckets,
780+
replicationFactor,
781+
new TabletServerInfo[] {
782+
new TabletServerInfo(0, "rack0"),
783+
new TabletServerInfo(1, "rack1"),
784+
new TabletServerInfo(2, "rack2")
785+
});
786+
long t1Id = metadataManager.createTable(t1, TEST_TABLE, tableAssignment, false);
787+
verifyTableCreated(t1Id, tableAssignment, nBuckets, replicationFactor);
788+
789+
// get the origin bucket leaderAndIsr
790+
Map<TableBucket, LeaderAndIsr> bucketLeaderAndIsrMap =
791+
new HashMap<>(
792+
waitValue(
793+
() -> fromCtx((ctx) -> Optional.of(ctx.bucketLeaderAndIsr())),
794+
Duration.ofMinutes(1),
795+
"leader not elected"));
796+
797+
// verify AdjustIsrReceivedEvent
798+
CompletableFuture<AdjustIsrResponse> response = new CompletableFuture<>();
799+
eventProcessor
800+
.getCoordinatorEventManager()
801+
.put(new AdjustIsrReceivedEvent(bucketLeaderAndIsrMap, response));
802+
803+
retryVerifyContext(
804+
ctx ->
805+
bucketLeaderAndIsrMap.forEach(
806+
(tableBucket, leaderAndIsr) ->
807+
assertThat(ctx.getBucketLeaderAndIsr(tableBucket))
808+
.contains(
809+
leaderAndIsr.newLeaderAndIsr(
810+
leaderAndIsr.leader(),
811+
leaderAndIsr.isr()))));
812+
813+
// verify the response
814+
AdjustIsrResponse adjustIsrResponse = response.get();
815+
Map<TableBucket, AdjustIsrResultForBucket> resultForBucketMap =
816+
getAdjustIsrResponseData(adjustIsrResponse);
817+
assertThat(resultForBucketMap.keySet())
818+
.containsAnyElementsOf(bucketLeaderAndIsrMap.keySet());
819+
assertThat(resultForBucketMap.values()).allMatch(AdjustIsrResultForBucket::succeeded);
820+
}
821+
764822
private CoordinatorEventProcessor buildCoordinatorEventProcessor() {
765823
return new CoordinatorEventProcessor(
766824
zookeeperClient,

‎fluss-server/src/test/java/com/alibaba/fluss/server/zk/ZooKeeperClientTest.java‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import java.util.Map;
5959
import java.util.Optional;
6060
import java.util.Set;
61+
import java.util.stream.Collectors;
6162

6263
import static com.alibaba.fluss.server.utils.TableAssignmentUtils.generateAssignment;
6364
import static org.assertj.core.api.Assertions.assertThat;
@@ -220,6 +221,45 @@ void testBatchCreateLeaderAndIsr() throws Exception {
220221
}
221222
}
222223

224+
@Test
225+
void testBatchUpdateLeaderAndIsr() throws Exception {
226+
int totalCount = 100;
227+
228+
// try to register bucket leaderAndIsr
229+
Map<TableBucket, LeaderAndIsr> leaderAndIsrList = new HashMap<>();
230+
for (int i = 0; i < totalCount; i++) {
231+
TableBucket tableBucket = new TableBucket(1, i);
232+
LeaderAndIsr leaderAndIsr =
233+
new LeaderAndIsr(i, 10, Arrays.asList(i + 1, i + 2, i + 3), 100, 1000);
234+
leaderAndIsrList.put(tableBucket, leaderAndIsr);
235+
zookeeperClient.registerLeaderAndIsr(tableBucket, leaderAndIsr);
236+
}
237+
238+
// try to batch update
239+
Map<TableBucket, LeaderAndIsr> updateLeaderAndIsrList =
240+
leaderAndIsrList.entrySet().stream()
241+
.collect(
242+
Collectors.toMap(
243+
Map.Entry::getKey,
244+
entry -> {
245+
LeaderAndIsr old = entry.getValue();
246+
return new LeaderAndIsr(
247+
old.leader() + 1,
248+
old.leaderEpoch() + 1,
249+
old.isr(),
250+
old.coordinatorEpoch() + 1,
251+
old.bucketEpoch() + 1);
252+
}));
253+
zookeeperClient.batchUpdateLeaderAndIsr(updateLeaderAndIsrList);
254+
for (Map.Entry<TableBucket, LeaderAndIsr> entry : updateLeaderAndIsrList.entrySet()) {
255+
TableBucket tableBucket = entry.getKey();
256+
LeaderAndIsr leaderAndIsr = entry.getValue();
257+
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket)).hasValue(leaderAndIsr);
258+
zookeeperClient.deleteLeaderAndIsr(tableBucket);
259+
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket)).isEmpty();
260+
}
261+
}
262+
223263
@Test
224264
void testTable() throws Exception {
225265
TablePath tablePath = TablePath.of("db", "tb");

0 commit comments

Comments
 (0)