Vladsz83 commented on code in PR #11361:
URL: https://github.com/apache/ignite/pull/11361#discussion_r1633241269
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java:
##########
@@ -3778,54 +3777,41 @@ private synchronized void scheduleNext() {
public synchronized void stop() {
stopping = true;
+ IgniteException err = new
IgniteException(SNP_NODE_STOPPING_ERR_MSG);
+
if (active != null)
- active.acceptException(new
IgniteException(SNP_NODE_STOPPING_ERR_MSG));
+ active.acceptException(err);
RemoteSnapshotFilesRecevier r;
while ((r = queue.poll()) != null)
- r.acceptException(new
IgniteException(SNP_NODE_STOPPING_ERR_MSG));
-
- Set<RemoteSnapshotFilesRecevier> futs = activeTasks();
- GridCompoundFuture<Void, Void> stopFut = new
GridCompoundFuture<>();
-
- try {
- for (IgniteInternalFuture<Void> fut : futs)
- stopFut.add(fut);
-
- stopFut.markInitialized().get();
- }
- catch (IgniteCheckedException e) {
- throw new IgniteException(e);
- }
+ r.acceptException(err);
}
/**
* @param nodeId A node left the cluster.
*/
- public void onNodeLeft(UUID nodeId) {
- Set<RemoteSnapshotFilesRecevier> futs = activeTasks();
+ public synchronized void onNodeLeft(UUID nodeId) {
+ if (active == null || active.isDone())
+ return;
+
ClusterTopologyCheckedException ex = new
ClusterTopologyCheckedException("The node from which a snapshot has been " +
"requested left the grid");
- futs.forEach(t -> {
+ if (active.rmtNodeId.equals(nodeId))
+ interruptActive(ex);
+
+ queue.forEach(t -> {
if (t.rmtNodeId.equals(nodeId))
t.acceptException(ex);
});
}
- /**
- * @return The set of currently scheduled tasks, some of them may be
already completed.
- */
- private Set<RemoteSnapshotFilesRecevier> activeTasks() {
- Set<RemoteSnapshotFilesRecevier> futs = new HashSet<>(queue);
-
- RemoteSnapshotFilesRecevier active0 = active;
-
- if (active0 != null)
- futs.add(active0);
+ /** Interrupts current active task (if present) to safely shedule
next. */
+ private void interruptActive(Exception e) {
+
cctx.kernalContext().io().interruptTransmissionReceiver(DFLT_INITIAL_SNAPSHOT_TOPIC,
e);
Review Comment:
Do we need this call? Or can we move it somewhere else? In `acceptException`
may be or to another stop/close handler?
--
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
