sashapolo commented on code in PR #3216:
URL: https://github.com/apache/ignite-3/pull/3216#discussion_r1494172795
##########
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:
Stopping index task removes the index as part of its job. This means such
tasks will cancel themselves, which looks weird to me. Is there a scenario when
an index can be removed not by a stopping task?
--
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]
