X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fcds-access-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Faccess%2Fclient%2FClientActorBehavior.java;h=3f8c11a9137ed3bc5cccda43b8a37f55d6f46f3d;hb=HEAD;hp=fa2e3b76d8a038497d57efd7e344498862717d06;hpb=634dfac8eead60f443bf75e749c70d1f2bb29198;p=controller.git diff --git a/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/ClientActorBehavior.java b/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/ClientActorBehavior.java index fa2e3b76d8..3f8c11a913 100644 --- a/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/ClientActorBehavior.java +++ b/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/ClientActorBehavior.java @@ -7,8 +7,8 @@ */ package org.opendaylight.controller.cluster.access.client; -import com.google.common.annotations.Beta; -import com.google.common.base.Preconditions; +import static java.util.Objects.requireNonNull; + import com.google.common.base.Stopwatch; import com.google.common.base.Verify; import java.util.Collection; @@ -17,9 +17,9 @@ import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import javax.annotation.Nonnull; -import javax.annotation.Nullable; -import javax.annotation.concurrent.GuardedBy; +import org.checkerframework.checker.lock.qual.Holding; +import org.eclipse.jdt.annotation.NonNull; +import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.controller.cluster.access.commands.NotLeaderException; import org.opendaylight.controller.cluster.access.commands.OutOfSequenceEnvelopeException; import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier; @@ -37,16 +37,14 @@ import org.opendaylight.controller.cluster.io.FileBackedOutputStreamFactory; import org.opendaylight.controller.cluster.messaging.MessageAssembler; import org.opendaylight.yangtools.concepts.Identifiable; import org.opendaylight.yangtools.concepts.Identifier; +import org.opendaylight.yangtools.concepts.Registration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.duration.FiniteDuration; /** * A behavior, which handles messages sent to a {@link AbstractClientActor}. - * - * @author Robert Varga */ -@Beta public abstract class ClientActorBehavior extends RecoveredClientActorBehavior implements Identifiable { /** @@ -60,7 +58,7 @@ public abstract class ClientActorBehavior extends * @param enqueuedEntries Previously-enqueued entries * @return A {@link ReconnectForwarder} to handle any straggler messages which arrive after this method returns. */ - @Nonnull ReconnectForwarder finishReconnect(@Nonnull Collection enqueuedEntries); + @NonNull ReconnectForwarder finishReconnect(@NonNull Collection enqueuedEntries); } private static final Logger LOG = LoggerFactory.getLogger(ClientActorBehavior.class); @@ -83,28 +81,41 @@ public abstract class ClientActorBehavior extends private final InversibleLock connectionsLock = new InversibleLock(); private final BackendInfoResolver resolver; private final MessageAssembler responseMessageAssembler; + private final Registration staleBackendInfoReg; - protected ClientActorBehavior(@Nonnull final ClientActorContext context, - @Nonnull final BackendInfoResolver resolver) { + protected ClientActorBehavior(final @NonNull ClientActorContext context, + final @NonNull BackendInfoResolver resolver) { super(context); - this.resolver = Preconditions.checkNotNull(resolver); + this.resolver = requireNonNull(resolver); final ClientActorConfig config = context.config(); responseMessageAssembler = MessageAssembler.builder().logContext(persistenceId()) .fileBackedStreamFactory(new FileBackedOutputStreamFactory(config.getFileBackedStreamingThreshold(), config.getTempFileDirectory())) .assembledMessageCallback((message, sender) -> context.self().tell(message, sender)).build(); + + staleBackendInfoReg = resolver.notifyWhenBackendInfoIsStale(shard -> { + context().executeInActor(behavior -> { + LOG.debug("BackendInfo for shard {} is now stale", shard); + final AbstractClientConnection conn = connections.get(shard); + if (conn instanceof ConnectedClientConnection) { + conn.reconnect(this, new BackendStaleException(shard)); + } + return behavior; + }); + }); } @Override - @Nonnull public final ClientIdentifier getIdentifier() { return context().getIdentifier(); } @Override public void close() { + super.close(); responseMessageAssembler.close(); + staleBackendInfoReg.close(); } /** @@ -137,12 +148,11 @@ public abstract class ClientActorBehavior extends return ((InternalCommand) command).execute(this); } - if (command instanceof SuccessEnvelope) { - return onRequestSuccess((SuccessEnvelope) command); + if (command instanceof SuccessEnvelope successEnvelope) { + return onRequestSuccess(successEnvelope); } - - if (command instanceof FailureEnvelope) { - return internalOnRequestFailure((FailureEnvelope) command); + if (command instanceof FailureEnvelope failureEnvelope) { + return internalOnRequestFailure(failureEnvelope); } if (MessageAssembler.isHandledMessage(command)) { @@ -159,10 +169,10 @@ public abstract class ClientActorBehavior extends } private static long extractCookie(final Identifier id) { - if (id instanceof TransactionIdentifier) { - return ((TransactionIdentifier) id).getHistoryId().getCookie(); - } else if (id instanceof LocalHistoryIdentifier) { - return ((LocalHistoryIdentifier) id).getCookie(); + if (id instanceof TransactionIdentifier transactionId) { + return transactionId.getHistoryId().getCookie(); + } else if (id instanceof LocalHistoryIdentifier historyId) { + return historyId.getCookie(); } else { throw new IllegalArgumentException("Unhandled identifier " + id); } @@ -200,7 +210,7 @@ public abstract class ClientActorBehavior extends * sessionId and if it does not match our current connection just ignore it. */ final Optional optBackend = conn.getBackendInfo(); - if (optBackend.isPresent() && optBackend.get().getSessionId() != command.getSessionId()) { + if (optBackend.isPresent() && optBackend.orElseThrow().getSessionId() != command.getSessionId()) { LOG.debug("{}: Mismatched current connection {} and envelope {}, ignoring response", persistenceId(), conn, command); return this; @@ -260,7 +270,7 @@ public abstract class ClientActorBehavior extends * * @param cause Failure cause */ - protected abstract void haltClient(@Nonnull Throwable cause); + protected abstract void haltClient(@NonNull Throwable cause); /** * Override this method to handle any command which is not handled by the base behavior. @@ -268,15 +278,14 @@ public abstract class ClientActorBehavior extends * @param command the command to process * @return Next behavior to use, null if this actor should shut down. */ - @Nullable - protected abstract ClientActorBehavior onCommand(@Nonnull Object command); + protected abstract @Nullable ClientActorBehavior onCommand(@NonNull Object command); /** * Override this method to provide a backend resolver instance. * * @return a backend resolver instance */ - protected final @Nonnull BackendInfoResolver resolver() { + protected final @NonNull BackendInfoResolver resolver() { return resolver; } @@ -287,8 +296,8 @@ public abstract class ClientActorBehavior extends * @param newConn New connection * @return ConnectionConnectCohort which will be used to complete the process of bringing the connection up. */ - @GuardedBy("connectionsLock") - @Nonnull protected abstract ConnectionConnectCohort connectionUp(@Nonnull ConnectedClientConnection newConn); + @Holding("connectionsLock") + protected abstract @NonNull ConnectionConnectCohort connectionUp(@NonNull ConnectedClientConnection newConn); private void backendConnectFinished(final Long shard, final AbstractClientConnection oldConn, final T backend, final Throwable failure) { @@ -313,8 +322,8 @@ public abstract class ClientActorBehavior extends LOG.error("{}: failed to resolve shard {}", persistenceId(), shard, failure); final RequestException cause; - if (failure instanceof RequestException) { - cause = (RequestException) failure; + if (failure instanceof RequestException requestException) { + cause = requestException; } else { cause = new RuntimeRequestException("Failed to resolve shard " + shard, failure); } @@ -406,7 +415,7 @@ public abstract class ClientActorBehavior extends final Long shard = oldConn.cookie(); LOG.info("{}: refreshing backend for shard {}", persistenceId(), shard); - resolver().refreshBackendInfo(shard, conn.getBackendInfo().get()).whenComplete( + resolver().refreshBackendInfo(shard, conn.getBackendInfo().orElseThrow()).whenComplete( (backend, failure) -> context().executeInActor(behavior -> { backendConnectFinished(shard, conn, backend, failure); return behavior; @@ -425,7 +434,8 @@ public abstract class ClientActorBehavior extends } private ConnectingClientConnection createConnection(final Long shard) { - final ConnectingClientConnection conn = new ConnectingClientConnection<>(context(), shard); + final ConnectingClientConnection conn = new ConnectingClientConnection<>(context(), shard, + resolver().resolveCookieName(shard)); resolveConnection(shard, conn); return conn; } @@ -437,4 +447,17 @@ public abstract class ClientActorBehavior extends return behavior; })); } + + private static class BackendStaleException extends RequestException { + private static final long serialVersionUID = 1L; + + BackendStaleException(final Long shard) { + super("Backend for shard " + shard + " is stale"); + } + + @Override + public boolean isRetriable() { + return false; + } + } }