vldpyatkov commented on code in PR #3422:
URL: https://github.com/apache/ignite-3/pull/3422#discussion_r1562366972
##########
modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngine.java:
##########
@@ -375,72 +362,61 @@ private CompletableFuture<Void>
recalculateAssignmentsAndScheduleRebalance(
return nullCompletedFuture();
}
- List<CatalogTableDescriptor> tableDescriptors =
findTablesByZoneId(zoneDescriptor.id(), catalogVersion, catalogService);
-
- return triggerPartitionsRebalanceForAllTables(
+ return triggerPartitionsRebalanceForZone(
causalityToken,
zoneDescriptor,
- dataNodes,
- tableDescriptors
+ dataNodes
);
});
}
- private CompletableFuture<Void> triggerPartitionsRebalanceForAllTables(
+ private CompletableFuture<Void> triggerPartitionsRebalanceForZone(
long revision,
CatalogZoneDescriptor zoneDescriptor,
- Set<String> dataNodes,
- List<CatalogTableDescriptor> tableDescriptors
+ Set<String> dataNodes
) {
- List<CompletableFuture<?>> tableFutures = new
ArrayList<>(tableDescriptors.size());
-
- for (CatalogTableDescriptor tableDescriptor : tableDescriptors) {
- CompletableFuture<?>[] partitionFutures =
RebalanceUtil.triggerAllTablePartitionsRebalance(
- tableDescriptor,
- zoneDescriptor,
- dataNodes,
- revision,
- metaStorageManager
- );
-
- // This set is used to deduplicate exceptions (if there is an
exception from upstream, for instance,
- // when reading from MetaStorage, it will be encountered by every
partition future) to avoid noise
- // in the logs.
- Set<Throwable> unwrappedCauses = ConcurrentHashMap.newKeySet();
-
- for (int partId = 0; partId < partitionFutures.length; partId++) {
- int finalPartId = partId;
-
- partitionFutures[partId].exceptionally(e -> {
- Throwable cause = ExceptionUtils.unwrapCause(e);
-
- if (unwrappedCauses.add(cause)) {
- // The exception is specific to this partition.
- LOG.error(
- "Exception on updating assignments for
[table={}, partition={}]",
- e,
- tableInfo(tableDescriptor), finalPartId
- );
- } else {
- // The exception is from upstream and not specific for
this partition, so don't log the partition index.
- LOG.error(
- "Exception on updating assignments for
[table={}]",
- e,
- tableInfo(tableDescriptor)
- );
- }
+ CompletableFuture<?>[] partitionFutures =
RebalanceUtil.triggerZonePartitionsRebalance(
+ zoneDescriptor,
+ dataNodes,
+ revision,
+ metaStorageManager
+ );
- return null;
- });
- }
+ // This set is used to deduplicate exceptions (if there is an
exception from upstream, for instance,
+ // when reading from MetaStorage, it will be encountered by every
partition future) to avoid noise
+ // in the logs.
+ Set<Throwable> unwrappedCauses = ConcurrentHashMap.newKeySet();
+
+ for (int partId = 0; partId < partitionFutures.length; partId++) {
+ int finalPartId = partId;
+
Review Comment:
finalPartId is needed to use it in the closure.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
