import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
import org.opendaylight.controller.cluster.datastore.messages.OnDemandShardState;
import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
-import org.opendaylight.controller.cluster.datastore.messages.PersistAbortTransactionPayload;
import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
-import org.opendaylight.controller.cluster.datastore.persisted.AbortTransactionPayload;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot.ShardSnapshot;
import org.opendaylight.controller.cluster.datastore.persisted.DisableTrackingPayload;
-import org.opendaylight.controller.cluster.datastore.persisted.PurgeTransactionPayload;
import org.opendaylight.controller.cluster.messaging.MessageAssembler;
import org.opendaylight.controller.cluster.messaging.MessageSlicer;
import org.opendaylight.controller.cluster.messaging.SliceOptions;
} else if (message instanceof DataTreeCohortActorRegistry.CohortRegistryCommand) {
store.processCohortRegistryCommand(getSender(),
(DataTreeCohortActorRegistry.CohortRegistryCommand) message);
- } else if (message instanceof PersistAbortTransactionPayload) {
- final TransactionIdentifier txId = ((PersistAbortTransactionPayload) message).getTransactionId();
- persistPayload(txId, AbortTransactionPayload.create(txId,
- datastoreContext.getInitialPayloadSerializedBufferCapacity()), true);
- persistPayload(txId, PurgeTransactionPayload.create(txId,
- datastoreContext.getInitialPayloadSerializedBufferCapacity()), false);
} else if (message instanceof MakeLeaderLocal) {
onMakeLeaderLocal();
} else if (RESUME_NEXT_PENDING_TRANSACTION.equals(message)) {
}
} else {
LOG.debug("{}: leader state for {} not found", persistenceId(), clientId);
+ knownFrontends.put(frontendId, new LeaderFrontendState.Disabled(persistenceId(), clientId,
+ getDataStore()));
}
}
}
if (isLeader()) {
final LocalHistoryIdentifier id = closeTransactionChain.getIdentifier();
askProtocolEncountered(id.getClientId());
-
- // FIXME: CONTROLLER-1628: stage purge once no transactions are present
- store.closeTransactionChain(id, null);
- store.purgeTransactionChain(id, null);
+ store.closeTransactionChain(id);
} else if (getLeader() != null) {
getLeader().forward(closeTransactionChain, getContext());
} else {
// Called on leader only
private void askProtocolEncountered(final ClientIdentifier clientId) {
- final LeaderFrontendState state = knownFrontends.get(clientId.getFrontendId());
- if (state instanceof LeaderFrontendState.Enabled) {
+ final FrontendIdentifier frontend = clientId.getFrontendId();
+ final LeaderFrontendState state = knownFrontends.get(frontend);
+ if (!(state instanceof LeaderFrontendState.Disabled)) {
LOG.debug("{}: encountered ask-based client {}, disabling transaction tracking", persistenceId(), clientId);
+ if (knownFrontends.isEmpty()) {
+ knownFrontends = new HashMap<>();
+ }
+ knownFrontends.put(frontend, new LeaderFrontendState.Disabled(persistenceId(), clientId, getDataStore()));
+
persistPayload(clientId, DisableTrackingPayload.create(clientId,
datastoreContext.getInitialPayloadSerializedBufferCapacity()), false);
}
paused = true;
// Tell-based protocol can replay transaction state, so it is safe to blow it up when we are paused.
- knownFrontends.values().forEach(LeaderFrontendState::retire);
- knownFrontends = ImmutableMap.of();
+ if (datastoreContext.isUseTellBasedProtocol()) {
+ knownFrontends.values().forEach(LeaderFrontendState::retire);
+ knownFrontends = ImmutableMap.of();
+ }
store.setRunOnPendingTransactionsComplete(operation);
}