Skip to content

Commit 7260389

Browse files
author
Liebing
committed
[server] Use batch operations of ZooKeeper to optimize the performance of updating LeaderAndIsr in tryProcessAdjustIsr
1 parent a71a337 commit 7260389

File tree

4 files changed

+148
-11
lines changed

4 files changed

+148
-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
@@ -93,6 +93,7 @@
9393

9494
import java.util.ArrayList;
9595
import java.util.Collections;
96+
import java.util.HashMap;
9697
import java.util.HashSet;
9798
import java.util.List;
9899
import java.util.Map;
@@ -889,6 +890,7 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
889890
// TODO verify leader epoch.
890891

891892
List<AdjustIsrResultForBucket> result = new ArrayList<>();
893+
Map<TableBucket, LeaderAndIsr> newLeaderAndIsrList = new HashMap<>();
892894
for (Map.Entry<TableBucket, LeaderAndIsr> entry : leaderAndIsrList.entrySet()) {
893895
TableBucket tableBucket = entry.getKey();
894896
LeaderAndIsr tryAdjustLeaderAndIsr = entry.getValue();
@@ -921,21 +923,37 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
921923
tryAdjustLeaderAndIsr.isr(),
922924
coordinatorContext.getCoordinatorEpoch(),
923925
currentLeaderAndIsr.bucketEpoch() + 1);
924-
try {
925-
zooKeeperClient.updateLeaderAndIsr(tableBucket, newLeaderAndIsr);
926-
} catch (Exception e) {
927-
LOG.error("Error when register leader and isr.", e);
928-
result.add(new AdjustIsrResultForBucket(tableBucket, ApiError.fromThrowable(e)));
926+
newLeaderAndIsrList.put(tableBucket, newLeaderAndIsr);
927+
}
928+
929+
try {
930+
zooKeeperClient.batchUpdateLeaderAndIsr(newLeaderAndIsrList);
931+
newLeaderAndIsrList.forEach(
932+
(tableBucket, newLeaderAndIsr) ->
933+
result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr)));
934+
} catch (Exception batchException) {
935+
LOG.error("Error when batch update leader and isr. Try one by one.", batchException);
936+
937+
for (Map.Entry<TableBucket, LeaderAndIsr> entry : newLeaderAndIsrList.entrySet()) {
938+
TableBucket tableBucket = entry.getKey();
939+
LeaderAndIsr newLeaderAndIsr = entry.getValue();
940+
try {
941+
zooKeeperClient.updateLeaderAndIsr(tableBucket, newLeaderAndIsr);
942+
} catch (Exception e) {
943+
LOG.error("Error when register leader and isr.", e);
944+
result.add(
945+
new AdjustIsrResultForBucket(tableBucket, ApiError.fromThrowable(e)));
946+
}
947+
// Successful return.
948+
result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr));
929949
}
950+
}
930951

931-
// update coordinator leader and isr cache.
932-
coordinatorContext.putBucketLeaderAndIsr(tableBucket, newLeaderAndIsr);
952+
// update coordinator leader and isr cache.
953+
newLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr);
933954

934-
// TODO update metadata for all alive tablet servers.
955+
// TODO update metadata for all alive tablet servers.
935956

936-
// Successful return.
937-
result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr));
938-
}
939957
return result;
940958
}
941959

‎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
@@ -301,6 +301,31 @@ public void updateLeaderAndIsr(TableBucket tableBucket, LeaderAndIsr leaderAndIs
301301
LOG.info("Updated {} for bucket {} in Zookeeper.", leaderAndIsr, tableBucket);
302302
}
303303

304+
public void batchUpdateLeaderAndIsr(Map<TableBucket, LeaderAndIsr> leaderAndIsrList)
305+
throws Exception {
306+
if (leaderAndIsrList.isEmpty()) {
307+
return;
308+
}
309+
310+
List<CuratorOp> ops = new ArrayList<>(leaderAndIsrList.size());
311+
for (Map.Entry<TableBucket, LeaderAndIsr> entry : leaderAndIsrList.entrySet()) {
312+
TableBucket tableBucket = entry.getKey();
313+
LeaderAndIsr leaderAndIsr = entry.getValue();
314+
315+
String path = LeaderAndIsrZNode.path(tableBucket);
316+
byte[] data = LeaderAndIsrZNode.encode(leaderAndIsr);
317+
CuratorOp updateOp = zkClient.transactionOp().setData().forPath(path, data);
318+
ops.add(updateOp);
319+
if (ops.size() == MAX_BATCH_SIZE) {
320+
zkClient.transaction().forOperations(ops);
321+
ops.clear();
322+
}
323+
}
324+
if (!ops.isEmpty()) {
325+
zkClient.transaction().forOperations(ops);
326+
}
327+
}
328+
304329
public void deleteLeaderAndIsr(TableBucket tableBucket) throws Exception {
305330
String path = LeaderAndIsrZNode.path(tableBucket);
306331
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: 36 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,41 @@ 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 leaderAndIsr = entry.getValue();
246+
return leaderAndIsr.newLeaderAndIsr(
247+
leaderAndIsr.leader(), leaderAndIsr.isr());
248+
}));
249+
zookeeperClient.batchUpdateLeaderAndIsr(updateLeaderAndIsrList);
250+
for (Map.Entry<TableBucket, LeaderAndIsr> entry : updateLeaderAndIsrList.entrySet()) {
251+
TableBucket tableBucket = entry.getKey();
252+
LeaderAndIsr leaderAndIsr = entry.getValue();
253+
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket)).hasValue(leaderAndIsr);
254+
zookeeperClient.deleteLeaderAndIsr(tableBucket);
255+
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket)).isEmpty();
256+
}
257+
}
258+
223259
@Test
224260
void testTable() throws Exception {
225261
TablePath tablePath = TablePath.of("db", "tb");

0 commit comments

Comments
 (0)