tkalkirill commented on code in PR #3216:
URL: https://github.com/apache/ignite-3/pull/3216#discussion_r1494398744
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java:
##########
@@ -146,39 +140,19 @@ public CompletableFuture<Void> start() {
startIndexes();
- catalogManager.listen(INDEX_CREATE, (parameters, exception) -> {
+ catalogService.listen(INDEX_CREATE, (parameters, exception) -> {
if (exception != null) {
Review Comment:
Maybe use `EventListener#fromConsumer`?
##########
modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java:
##########
@@ -110,77 +106,66 @@ public void close() throws Exception {
}
private void addListeners() {
- catalogService.listen(CatalogEvent.INDEX_CREATE,
adaptToEventListener(this::onIndexCreated));
+ catalogService.listen(CatalogEvent.INDEX_CREATE,
EventListener.fromConsumer(this::onIndexCreated));
- catalogService.listen(CatalogEvent.INDEX_STOPPING,
adaptToEventListener(this::onIndexDropped));
+ catalogService.listen(CatalogEvent.INDEX_STOPPING,
EventListener.fromConsumer(this::onIndexDropped));
- catalogService.listen(CatalogEvent.INDEX_REMOVED,
adaptToEventListener(this::onIndexRemoved));
+ catalogService.listen(CatalogEvent.INDEX_REMOVED,
EventListener.fromConsumer(this::onIndexRemoved));
- placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED,
adaptToEventListener(this::onPrimaryReplicaElected));
- }
-
- /** Wraps a given callback into an EventListener. */
- private <T extends EventParameters> EventListener<T>
adaptToEventListener(Consumer<T> action) {
- return (parameters, exception) -> {
- if (exception != null) {
- return failedFuture(exception);
- }
-
- inBusyLock(busyLock, () -> action.accept(parameters));
-
- return falseCompletedFuture();
- };
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED,
EventListener.fromConsumer(this::onPrimaryReplicaElected));
}
private void onIndexCreated(CreateIndexEventParameters parameters) {
- CatalogIndexDescriptor indexDescriptor = parameters.indexDescriptor();
+ inBusyLock(busyLock, () -> {
+ CatalogIndexDescriptor indexDescriptor =
parameters.indexDescriptor();
- if
(localNodeIsPrimaryReplicaForTableIds.contains(indexDescriptor.tableId())) {
- // Schedule building the index only if the local node is the
primary replica for the 0 partition of the table for which the
- // index was created.
-
changeIndexStatusTaskScheduler.scheduleStartBuildingTask(parameters.indexDescriptor());
- }
+ if
(localNodeIsPrimaryReplicaForTableIds.contains(indexDescriptor.tableId())) {
+ // Schedule building the index only if the local node is the
primary replica for the 0 partition of the table for which the
+ // index was created.
+
changeIndexStatusTaskScheduler.scheduleStartBuildingTask(parameters.indexDescriptor());
+ }
+ });
}
private void onIndexDropped(StoppingIndexEventParameters parameters) {
- CatalogIndexDescriptor indexDescriptor =
catalogService.index(parameters.indexId(), parameters.catalogVersion());
+ inBusyLock(busyLock, () -> {
+ CatalogIndexDescriptor indexDescriptor =
catalogService.index(parameters.indexId(), parameters.catalogVersion());
- assert indexDescriptor != null : parameters.indexId();
+ assert indexDescriptor != null : parameters.indexId();
- if
(localNodeIsPrimaryReplicaForTableIds.contains(indexDescriptor.tableId())) {
- // Schedule index removal only if the local node is the primary
replica for the 0 partition of the table for which the
- // index was dropped.
-
changeIndexStatusTaskScheduler.scheduleRemoveIndexTask(indexDescriptor);
- }
+ if
(localNodeIsPrimaryReplicaForTableIds.contains(indexDescriptor.tableId())) {
+ // Schedule index removal only if the local node is the
primary replica for the 0 partition of the table for which the
+ // index was dropped.
+
changeIndexStatusTaskScheduler.scheduleRemoveIndexTask(indexDescriptor);
+ }
+ });
}
private void onIndexRemoved(RemoveIndexEventParameters parameters) {
- CatalogIndexDescriptor indexDescriptor =
catalogService.index(parameters.indexId(), parameters.catalogVersion() - 1);
-
- assert indexDescriptor != null : parameters.indexId();
-
- changeIndexStatusTaskScheduler.stopTask(indexDescriptor);
+ inBusyLock(busyLock, () ->
changeIndexStatusTaskScheduler.stopStartBuildingTask(parameters.indexId()));
Review Comment:
My mistake.
##########
modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java:
##########
@@ -1390,7 +1389,7 @@ private void
registerCatalogEventListenersOnStartManagerBusy() {
catalogManager.listen(ZONE_DROP, (parameters, exception) ->
inBusyLock(busyLock, () -> {
assert exception == null : parameters;
- return onDropZoneBusy((DropZoneEventParameters)
parameters).thenCompose((ignored) -> falseCompletedFuture());
+ return onDropZoneBusy((DropZoneEventParameters)
parameters).thenApply((ignored) -> false);
Review Comment:
I meant that these changes, in theory, do not affect your changes, but oh
well.
##########
modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java:
##########
@@ -110,77 +106,66 @@ public void close() throws Exception {
}
private void addListeners() {
- catalogService.listen(CatalogEvent.INDEX_CREATE,
adaptToEventListener(this::onIndexCreated));
+ catalogService.listen(CatalogEvent.INDEX_CREATE,
EventListener.fromConsumer(this::onIndexCreated));
Review Comment:
My mistake.
--
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]
