ivanzlenko commented on code in PR #5592:
URL: https://github.com/apache/ignite-3/pull/5592#discussion_r2036691090
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2493,16 +2498,13 @@ private WatchListener
createAssignmentsSwitchRebalanceListener() {
long assignmentsTimestamp = assignments.timestamp();
- return
waitForMetadataCompleteness(assignmentsTimestamp)
- .thenCompose(unused ->
inBusyLockAsync(busyLock, () -> {
- int catalogVersion =
catalogService.activeCatalogVersion(assignmentsTimestamp);
+ return
reliableCatalogVersions.safeReliableCatalogFor(hybridTimestamp(assignmentsTimestamp))
+ .thenCompose(catalog ->
inBusyLockAsync(busyLock, () -> {
+ CatalogTableDescriptor tableDescriptor =
getTableDescriptor(replicaGrpId.tableId(), catalog);
- CatalogTableDescriptor tableDescriptor =
-
getTableDescriptor(replicaGrpId.tableId(), catalogVersion);
+ CatalogZoneDescriptor zoneDescriptor =
getZoneDescriptor(tableDescriptor, catalog);
- CatalogZoneDescriptor zoneDescriptor =
getZoneDescriptor(tableDescriptor, catalogVersion);
-
- return
distributionZoneManager.dataNodes(zoneDescriptor.updateTimestamp(),
catalogVersion,
+ return
distributionZoneManager.dataNodes(zoneDescriptor.updateTimestamp(),
catalog.version(),
tableDescriptor.zoneId())
.thenCompose(dataNodes ->
RebalanceUtilEx.handleReduceChanged(
metaStorageMgr,
Review Comment:
Let's move it to the new method as well
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -1392,26 +1396,27 @@ private CompletableFuture<Set<Assignment>>
calculateAssignments(
TablePartitionId tablePartitionId,
Long assignmentsTimestamp
) {
- return
orStopManagerFuture(waitForMetadataCompleteness(assignmentsTimestamp).thenCompose(unused
-> {
- int catalogVersion =
catalogService.activeCatalogVersion(assignmentsTimestamp);
-
- CatalogTableDescriptor tableDescriptor =
getTableDescriptor(tablePartitionId.tableId(), catalogVersion);
-
- CatalogZoneDescriptor zoneDescriptor =
getZoneDescriptor(tableDescriptor, catalogVersion);
+ CompletableFuture<Set<Assignment>> assignmentsFuture =
+
reliableCatalogVersions.safeReliableCatalogFor(hybridTimestamp(assignmentsTimestamp))
+ .thenCompose(catalog -> {
+ CatalogTableDescriptor tableDescriptor =
getTableDescriptor(tablePartitionId.tableId(), catalog);
+ CatalogZoneDescriptor zoneDescriptor =
getZoneDescriptor(tableDescriptor, catalog);
+
+ return distributionZoneManager.dataNodes(
+ zoneDescriptor.updateTimestamp(),
+ catalog.version(),
+ tableDescriptor.zoneId()
+ ).thenApply(dataNodes ->
Review Comment:
Let's move it into a separate method
--
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: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
