Fix RaftStorage actor confinement violation

We are invoking a callback from a background thread. This should be
routed via executeInSelf() to ensure we execute in proper context.

JIRA: CONTROLLER-2134
Change-Id: I92a40ebd7ed170ae7a77d7923eedefe285978692
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>

diff --git a/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/PekkoRaftStorage.java b/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/PekkoRaftStorage.java
index 84c6ef18828255e431bd5ee146b250e86715856e..c83a9d17d8ce0b5f767c4ca4bbb6479a93ffdb50 100644 (file)
--- a/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/PekkoRaftStorage.java
+++ b/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/PekkoRaftStorage.java
@@ -29,7 +29,7 @@ final class PekkoRaftStorage extends EnabledRaftStorage {
     private final RaftActor actor;
 
     PekkoRaftStorage(final RaftActor actor, final CompressionSupport compression, final Configuration streamConfig) {
-        super(compression, streamConfig);
+        super(actor, compression, streamConfig);
         this.actor = requireNonNull(actor);
     }
 

diff --git a/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/DisabledRaftStorage.java b/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/DisabledRaftStorage.java
index 7b77ae17027b2e86feb0169eecfd588bbce8935f..fe5cc973c19c3bdae33e4d45d7ff2bac20340082 100644 (file)
--- a/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/DisabledRaftStorage.java
+++ b/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/DisabledRaftStorage.java
@@ -43,14 +43,12 @@ public final class DisabledRaftStorage extends RaftStorage implements ImmediateD
     }
 
     private final String memberId;
-    private final ExecuteInSelfActor actor;
     private final ActorRef actorRef;
 
-    public DisabledRaftStorage(final String memberId, final ExecuteInSelfActor actor, final ActorRef actorRef,
+    public DisabledRaftStorage(final String memberId, final ExecuteInSelfActor executeInSelf, final ActorRef actorRef,
             final CompressionSupport compression, final Configuration streamConfig) {
-        super(compression, streamConfig);
+        super(executeInSelf, compression, streamConfig);
         this.memberId = requireNonNull(memberId);
-        this.actor = requireNonNull(actor);
         this.actorRef = requireNonNull(actorRef);
     }
 
@@ -71,7 +69,7 @@ public final class DisabledRaftStorage extends RaftStorage implements ImmediateD
 
     @Override
     public ExecuteInSelfActor actor() {
-        return actor;
+        return executeInSelf;
     }
 
     @Override

diff --git a/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/EnabledRaftStorage.java b/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/EnabledRaftStorage.java
index 3fb7fe6d3fdcbaee2001e16330e9dd57d645e28d..7504fdc4677346cd7b325c7998f900a7c989c70b 100644 (file)
--- a/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/EnabledRaftStorage.java
+++ b/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/EnabledRaftStorage.java
@@ -8,6 +8,7 @@
 package org.opendaylight.controller.cluster.raft.spi;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.opendaylight.controller.cluster.common.actor.ExecuteInSelfActor;
 import org.opendaylight.controller.cluster.raft.RaftActor;
 import org.opendaylight.raft.spi.CompressionSupport;
 import org.opendaylight.raft.spi.FileBackedOutputStream.Configuration;
@@ -17,8 +18,9 @@ import org.opendaylight.raft.spi.FileBackedOutputStream.Configuration;
  */
 @NonNullByDefault
 public abstract non-sealed class EnabledRaftStorage extends RaftStorage {
-    protected EnabledRaftStorage(final CompressionSupport compression, final Configuration streamConfig) {
-        super(compression, streamConfig);
+    protected EnabledRaftStorage(final ExecuteInSelfActor executeInSelf, final CompressionSupport compression,
+            final Configuration streamConfig) {
+        super(executeInSelf, compression, streamConfig);
     }
 
     @Override

diff --git a/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/RaftStorage.java b/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/RaftStorage.java
index 10fe4d97b302d850e0421a308facc384c634a968..8449767465378674437ba7e49d64e8392a37725d 100644 (file)
--- a/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/RaftStorage.java
+++ b/raft/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/spi/RaftStorage.java
@@ -20,6 +20,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.BiConsumer;
 import org.eclipse.jdt.annotation.NonNull;
+import org.opendaylight.controller.cluster.common.actor.ExecuteInSelfActor;
 import org.opendaylight.raft.spi.CompressionSupport;
 import org.opendaylight.raft.spi.FileBackedOutputStream;
 import org.opendaylight.raft.spi.FileBackedOutputStream.Configuration;
@@ -51,10 +52,10 @@ public abstract sealed class RaftStorage implements DataPersistenceProvider
             try {
                 result = compute();
             } catch (Exception e) {
-                callback.accept(null, e);
+                executeInSelf.executeInSelf(() -> callback.accept(null, e));
                 return;
             }
-            callback.accept(result, null);
+            executeInSelf.executeInSelf(() -> callback.accept(result, null));
         }
 
         abstract @NonNull T compute() throws Exception;
@@ -86,9 +87,13 @@ public abstract sealed class RaftStorage implements DataPersistenceProvider
     private final @NonNull CompressionSupport compression;
     private final @NonNull Configuration streamConfig;
 
+    protected final @NonNull ExecuteInSelfActor executeInSelf;
+
     private ExecutorService executor;
 
-    protected RaftStorage(final CompressionSupport compression, final Configuration streamConfig) {
+    protected RaftStorage(final ExecuteInSelfActor executeInSelf, final CompressionSupport compression,
+            final Configuration streamConfig) {
+        this.executeInSelf = requireNonNull(executeInSelf);
         this.compression = requireNonNull(compression);
         this.streamConfig = requireNonNull(streamConfig);
     }