<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-actor_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-actor-typed_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-cluster_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-cluster-typed_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-osgi_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-persistence_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-protobuf_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-remote_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-slf4j_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-stream_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
</dependency>
</dependencies>
akka.cluster.split-brain-resolver.lease-majority {
lease-implementation = ""
+ # The recommended format for the lease name is "<service-name>-akka-sbr".
+ # When lease-name is not defined, the name will be set to "<actor-system-name>-akka-sbr"
+ lease-name = ""
+
# This delay is used on the minority side before trying to acquire the lease,
# as an best effort to try to keep the majority side.
acquire-lease-delay-for-minority = 2s
+ # Release the lease after this duration.
+ release-after = 40s
+
# If the 'role' is defined the majority/minority is based only on members with that 'role'.
role = ""
}
call-timeout = 20s
reset-timeout = 60s
}
+
+ # Set this to true if successful loading of snapshot is not necessary.
+ # This can be useful when it is alright to ignore snapshot in case of
+ # for example deserialization errors. When snapshot loading fails it will instead
+ # recover by replaying all events.
+ # Don't set to true if events are deleted because that would
+ # result in wrong recovered state if snapshot load fails.
+ snapshot-is-optional = false
+
}
fsm {
artery {
# Disable artery with this flag
- enabled = off
+ enabled = on
# Select the underlying transport implementation.
#
# Possible values: aeron-udp, tcp, tls-tcp
# See https://doc.akka.io/docs/akka/current/remoting-artery.html#selecting-a-transport for the tradeoffs
# for each transport
- transport = aeron-udp
+ transport = tcp
# Canonical address is the address other clients should connect to.
# Artery transport will expect messages to this address.
# slightly more bytes than this limit (at most one element more). It can be set to 0
# to disable the usage of the buffer.
write-buffer-size = 16 KiB
+
+ # In addition to the buffering described for property write-buffer-size, try to collect
+ # more consecutive writes from the upstream stream producers.
+ #
+ # The rationale is to increase write efficiency by avoiding separate small
+ # writes to the network which is expensive to do. Merging those writes together
+ # (up to `write-buffer-size`) improves throughput for small writes.
+ #
+ # The idea is that a running stream may produce multiple small writes consecutively
+ # in one go without waiting for any external input. To probe the stream for
+ # data, this features delays sending a write immediately by probing the stream
+ # for more writes. This works by rescheduling the TCP connection stage via the
+ # actor mailbox of the underlying actor. Thus, before the stage is reactivated
+ # the upstream gets another opportunity to emit writes.
+ #
+ # When the stage is reactivated and if new writes are detected another round-trip
+ # is scheduled. The loop repeats until either the number of round trips given in this
+ # setting is reached, the buffer reaches `write-buffer-size`, or no new writes
+ # were detected during the last round-trip.
+ #
+ # This mechanism ensures that a write is guaranteed to be sent when the remaining stream
+ # becomes idle waiting for external signals.
+ #
+ # In most cases, the extra latency this mechanism introduces should be negligible,
+ # but depending on the stream setup it may introduce a noticeable delay,
+ # if the upstream continuously produces small amounts of writes in a
+ # blocking (CPU-bound) way.
+ #
+ # In that case, the feature can either be disabled, or the producing CPU-bound
+ # work can be taken off-stream to avoid excessive delays (e.g. using `mapAsync` instead of `map`).
+ #
+ # A value of 0 disables this feature.
+ coalesce-writes = 10
}
# Time to wait for async materializer creation before throwing an exception
<dependency>
<groupId>org.scalatestplus</groupId>
<artifactId>junit-4-13_2.13</artifactId>
- <version>3.1.4.0</version>
+ <version>3.2.5.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-testkit_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
<scope>test</scope>
<exclusions>
<exclusion>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-actor-testkit-typed_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
<scope>test</scope>
<exclusions>
<exclusion>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-persistence-tck_2.13</artifactId>
- <version>2.6.12</version>
+ <version>2.6.16</version>
<scope>test</scope>
<exclusions>
<exclusion>
<dependency>
<groupId>org.agrona</groupId>
<artifactId>agrona</artifactId>
- <version>1.8.0</version>
+ <version>1.9.0</version>
</dependency>
<dependency>
<groupId>io.aeron</groupId>
<artifactId>aeron-client</artifactId>
- <version>1.31.1</version>
+ <version>1.32.0</version>
</dependency>
<dependency>
<groupId>io.aeron</groupId>
<artifactId>aeron-driver</artifactId>
- <version>1.31.1</version>
+ <version>1.32.0</version>
</dependency>
</dependencies>
</dependencyManagement>
<link>http://google.github.io/truth/api/1.0.1/</link>
<link>http://www.slf4j.org/apidocs/</link>
<link>https://google.github.io/guava/releases/29.0-jre/api/docs/</link>
- <link>http://doc.akka.io/japi/akka/2.6.12/</link>
+ <link>http://doc.akka.io/japi/akka/2.6.16/</link>
<link>http://netty.io/4.1/api/</link>
<link>https://commons.apache.org/proper/commons-lang/javadocs/api-2.6/</link>
<link>https://commons.apache.org/proper/commons-lang/javadocs/api-3.9/</link>
<feature>odl-controller-scala</feature>
<bundle>mvn:com.typesafe/config/1.4.0</bundle>
<bundle>mvn:com.typesafe/ssl-config-core_2.13/0.4.2</bundle>
- <bundle>mvn:io.aeron/aeron-client/1.31.1</bundle>
- <bundle>mvn:io.aeron/aeron-driver/1.31.1</bundle>
+ <bundle>mvn:io.aeron/aeron-client/1.32.0</bundle>
+ <bundle>mvn:io.aeron/aeron-driver/1.32.0</bundle>
<bundle>mvn:io.netty/netty/3.10.6.Final</bundle>
- <bundle>mvn:org.agrona/agrona/1.8.0</bundle>
+ <bundle>mvn:org.agrona/agrona/1.9.0</bundle>
<bundle>mvn:org.opendaylight.controller/repackaged-akka/${project.version}</bundle>
<bundle>mvn:org.reactivestreams/reactive-streams/1.0.3</bundle>
<feature>wrap</feature>
package org.opendaylight.controller.cluster.datastore;
+import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.mock;
import static org.mockito.MockitoAnnotations.initMocks;
import akka.actor.PoisonPill;
import akka.actor.Props;
import com.google.common.util.concurrent.SettableFuture;
+import java.time.Duration;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Before;
InMemorySnapshotStore.clear();
mockShardActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ await().atMost(Duration.ofSeconds(10)).until(mockShardActor::isTerminated);
mockShardActor = null;
actorFactory.close();
Code Review / controller.git / commitdiff