Skip to content

Commit 0cb0d4c

Browse files
vyazelenkovyazelenko
committed
[Java] Use unique session-id when creating live log recording and replaying data from the Cluster to ensure that the old replay data is not being picked up upon restart/reset.
(cherry picked from commit 4d167e9)
1 parent 05475a6 commit 0cb0d4c

3 files changed

Lines changed: 77 additions & 28 deletions

File tree

‎aeron-cluster/src/main/java/io/aeron/cluster/ClusterBackupAgent.java‎

Lines changed: 27 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import io.aeron.cluster.service.ClusterMarkFile;
4141
import io.aeron.exceptions.TimeoutException;
4242
import io.aeron.logbuffer.Header;
43+
import org.agrona.BitUtil;
4344
import org.agrona.CloseHelper;
4445
import org.agrona.DirectBuffer;
4546
import org.agrona.ErrorHandler;
@@ -55,11 +56,7 @@
5556
import java.util.concurrent.TimeUnit;
5657

5758
import static io.aeron.Aeron.NULL_VALUE;
58-
import static io.aeron.CommonContext.CONTROL_MODE_RESPONSE;
59-
import static io.aeron.CommonContext.ENDPOINT_PARAM_NAME;
60-
import static io.aeron.CommonContext.MDC_CONTROL_MODE_PARAM_NAME;
61-
import static io.aeron.CommonContext.MDC_CONTROL_PARAM_NAME;
62-
import static io.aeron.CommonContext.TAGS_PARAM_NAME;
59+
import static io.aeron.CommonContext.*;
6360
import static io.aeron.archive.client.AeronArchive.NULL_LENGTH;
6461
import static io.aeron.archive.client.AeronArchive.NULL_POSITION;
6562
import static io.aeron.archive.client.AeronArchive.NULL_TIMESTAMP;
@@ -142,7 +139,8 @@ public final class ClusterBackupAgent implements Agent
142139
private long liveLogRecordingId = NULL_VALUE;
143140
private long liveLogReplaySessionId = NULL_VALUE;
144141
private int leaderCommitPositionCounterId = NULL_VALUE;
145-
private int liveLogRecCounterId = NULL_COUNTER_ID;
142+
private int liveLogRecordingCounterId = NULL_COUNTER_ID;
143+
private int liveLogRecordingSessionId = NULL_VALUE;
146144

147145
ClusterBackupAgent(final ClusterBackup.Context ctx)
148146
{
@@ -336,8 +334,9 @@ private void reset()
336334
logSupplierMember = null;
337335
leaderLogEntry = null;
338336
leaderLastTermEntry = null;
339-
liveLogRecCounterId = NULL_COUNTER_ID;
337+
liveLogRecordingCounterId = NULL_COUNTER_ID;
340338
liveLogRecordingId = NULL_VALUE;
339+
liveLogRecordingSessionId = NULL_VALUE;
341340

342341
snapshotsToRetrieve.clear();
343342
snapshotsRetrieved.clear();
@@ -387,7 +386,7 @@ private void reset()
387386

388387
private void onUnavailableCounter(final CountersReader counters, final long registrationId, final int counterId)
389388
{
390-
if (counterId == liveLogRecCounterId)
389+
if (counterId == liveLogRecordingCounterId)
391390
{
392391
if (null != eventsListener)
393392
{
@@ -758,6 +757,11 @@ private int liveLogRecord(final long nowMs)
758757

759758
if (NULL_VALUE == liveLogRecordingSubscriptionId)
760759
{
760+
if (NULL_VALUE == liveLogRecordingSessionId)
761+
{
762+
liveLogRecordingSessionId = BitUtil.generateRandomisedId();
763+
}
764+
761765
final String catchupEndpoint = ctx.catchupEndpoint();
762766
if (catchupEndpoint.endsWith(":0"))
763767
{
@@ -766,10 +770,11 @@ private int liveLogRecord(final long nowMs)
766770
final ChannelUri channelUri = ChannelUri.parse(ctx.catchupChannel());
767771
channelUri.remove(ENDPOINT_PARAM_NAME);
768772
channelUri.put(TAGS_PARAM_NAME, aeron.nextCorrelationId() + "," + aeron.nextCorrelationId());
773+
channelUri.put(SESSION_ID_PARAM_NAME, Integer.toString(liveLogRecordingSessionId));
769774
recordingChannel = channelUri.toString();
770775

771-
final String channel = recordingChannel + "|endpoint=" + catchupEndpoint;
772-
recordingSubscription = aeron.addSubscription(channel, ctx.logStreamId());
776+
channelUri.put(ENDPOINT_PARAM_NAME, catchupEndpoint);
777+
recordingSubscription = aeron.addSubscription(channelUri.toString(), ctx.logStreamId());
773778
timeOfLastProgressMs = nowMs;
774779
return 1;
775780
}
@@ -784,6 +789,7 @@ private int liveLogRecord(final long nowMs)
784789
final ChannelUri channelUri = ChannelUri.parse(ctx.catchupChannel());
785790
channelUri.put(ENDPOINT_PARAM_NAME, catchupEndpoint);
786791
channelUri.replaceEndpointWildcardPort(resolvedEndpoint);
792+
channelUri.put(SESSION_ID_PARAM_NAME, Integer.toString(liveLogRecordingSessionId));
787793

788794
replayChannel = channelUri.toString();
789795
}
@@ -792,6 +798,7 @@ private int liveLogRecord(final long nowMs)
792798
{
793799
final ChannelUri channelUri = ChannelUri.parse(ctx.catchupChannel());
794800
channelUri.put(ENDPOINT_PARAM_NAME, catchupEndpoint);
801+
channelUri.put(SESSION_ID_PARAM_NAME, Integer.toString(liveLogRecordingSessionId));
795802
replayChannel = channelUri.toString();
796803
recordingChannel = replayChannel;
797804
}
@@ -849,16 +856,16 @@ else if (NULL_VALUE == liveLogReplaySessionId)
849856
timeOfLastProgressMs = nowMs;
850857
}
851858
}
852-
else if (NULL_COUNTER_ID == liveLogRecCounterId)
859+
else if (NULL_COUNTER_ID == liveLogRecordingCounterId)
853860
{
854861
final CountersReader countersReader = aeron.countersReader();
855862

856-
liveLogRecCounterId = RecordingPos.findCounterIdBySession(
863+
liveLogRecordingCounterId = RecordingPos.findCounterIdBySession(
857864
countersReader, (int)liveLogReplaySessionId, backupArchive.archiveId());
858-
if (NULL_COUNTER_ID != liveLogRecCounterId)
865+
if (NULL_COUNTER_ID != liveLogRecordingCounterId)
859866
{
860-
liveLogPositionCounter.setOrdered(countersReader.getCounterValue(liveLogRecCounterId));
861-
liveLogRecordingId = RecordingPos.getRecordingId(countersReader, liveLogRecCounterId);
867+
liveLogPositionCounter.setOrdered(countersReader.getCounterValue(liveLogRecordingCounterId));
868+
liveLogRecordingId = RecordingPos.getRecordingId(countersReader, liveLogRecordingCounterId);
862869
timeOfLastBackupQueryMs = nowMs;
863870
timeOfLastProgressMs = nowMs;
864871
state(UPDATE_RECORDING_LOG, nowMs);
@@ -959,15 +966,15 @@ private int backingUp(final long nowMs)
959966
workCount += 1;
960967
}
961968

962-
if (NULL_COUNTER_ID != liveLogRecCounterId)
969+
if (NULL_COUNTER_ID != liveLogRecordingCounterId)
963970
{
964-
final long liveLogPosition = aeron.countersReader().getCounterValue(liveLogRecCounterId);
971+
final long liveLogPosition = aeron.countersReader().getCounterValue(liveLogRecordingCounterId);
965972

966973
if (liveLogPositionCounter.proposeMaxOrdered(liveLogPosition))
967974
{
968975
if (null != eventsListener)
969976
{
970-
eventsListener.onLiveLogProgress(liveLogRecordingId, liveLogRecCounterId, liveLogPosition);
977+
eventsListener.onLiveLogProgress(liveLogRecordingId, liveLogRecordingCounterId, liveLogPosition);
971978
}
972979

973980
workCount += 1;
@@ -1086,7 +1093,8 @@ private long startLogRecording()
10861093

10871094
private boolean hasProgressStalled(final long nowMs)
10881095
{
1089-
return (NULL_COUNTER_ID == liveLogRecCounterId) && (nowMs > (timeOfLastProgressMs + backupProgressTimeoutMs));
1096+
return (NULL_COUNTER_ID == liveLogRecordingCounterId) &&
1097+
(nowMs > (timeOfLastProgressMs + backupProgressTimeoutMs));
10901098
}
10911099

10921100
private long replayStartPosition(final RecordingLog.Entry lastTerm)

‎aeron-system-tests/src/test/java/io/aeron/cluster/ClusterBackupTest.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import io.aeron.samples.archive.SampleAuthorisationService;
2121
import io.aeron.security.AuthenticatorSupplier;
2222
import io.aeron.security.AuthorisationServiceSupplier;
23+
import io.aeron.security.NullCredentialsSupplier;
2324
import io.aeron.test.EventLogExtension;
2425
import io.aeron.test.InterruptAfter;
2526
import io.aeron.test.InterruptingTestCallback;
@@ -37,6 +38,7 @@
3738
import org.junit.jupiter.api.extension.RegisterExtension;
3839
import org.junit.jupiter.params.ParameterizedTest;
3940
import org.junit.jupiter.params.provider.EnumSource;
41+
import org.junit.jupiter.params.provider.ValueSource;
4042

4143
import static io.aeron.cluster.ClusterBackup.Configuration.ReplayStart.LATEST_SNAPSHOT;
4244
import static io.aeron.test.SystemTestWatcher.UNKNOWN_HOST_FILTER;
@@ -627,6 +629,36 @@ void shouldBackupClusterAndJoinLive()
627629
cluster.awaitServiceMessageCount(node, 200);
628630
}
629631

632+
@ParameterizedTest
633+
@ValueSource(ints = { 0, 8833 })
634+
@InterruptAfter(60)
635+
void shouldResumeBackupIfStopped(final int catchupPort)
636+
{
637+
final TestCluster cluster = aCluster().withStaticNodes(3).start();
638+
systemTestWatcher.cluster(cluster);
639+
640+
cluster.connectClient();
641+
final int initialMessageCount = 100_000; // minimum number of messages to trigger the bug
642+
cluster.sendMessages(initialMessageCount);
643+
cluster.awaitServicesMessageCount(initialMessageCount);
644+
645+
final TestBackupNode backupNode = cluster.startClusterBackupNode(
646+
true, new NullCredentialsSupplier(), ClusterBackup.SourceType.FOLLOWER, catchupPort);
647+
cluster.awaitBackupLiveLogPosition(1);
648+
backupNode.close();
649+
650+
final int delta = 1000;
651+
cluster.sendMessages(delta);
652+
653+
cluster.awaitServicesMessageCount(initialMessageCount + delta);
654+
cluster.awaitResponseMessageCount(initialMessageCount + delta);
655+
656+
cluster.startClusterBackupNode(
657+
false, new NullCredentialsSupplier(), ClusterBackup.SourceType.FOLLOWER, catchupPort);
658+
cluster.awaitBackupState(ClusterBackup.State.BACKING_UP);
659+
cluster.awaitBackupLiveLogPosition(cluster.findLeader().service().cluster().logPosition());
660+
}
661+
630662
private static void awaitErrorLogged(final TestBackupNode testBackupNode, final String expectedErrorMessage)
631663
{
632664
final AtomicBuffer atomicBuffer = testBackupNode.clusterBackupErrorLog();

‎aeron-test-support/src/main/java/io/aeron/test/cluster/TestCluster.java‎

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -355,6 +355,15 @@ public TestBackupNode startClusterBackupNode(
355355
final boolean cleanStart,
356356
final CredentialsSupplier credentialsSupplier,
357357
final ClusterBackup.SourceType sourceType)
358+
{
359+
return startClusterBackupNode(cleanStart, credentialsSupplier, sourceType, 0);
360+
}
361+
362+
public TestBackupNode startClusterBackupNode(
363+
final boolean cleanStart,
364+
final CredentialsSupplier credentialsSupplier,
365+
final ClusterBackup.SourceType sourceType,
366+
final int catchupEndpointPort)
358367
{
359368
final int index = staticMemberCount;
360369
final String baseDirName = clusterBaseDir + "-" + index;
@@ -397,17 +406,17 @@ public TestBackupNode startClusterBackupNode(
397406
.clusterConsensusEndpoints(clusterConsensusEndpoints)
398407
.consensusChannel(consensusChannelUri.toString())
399408
.clusterBackupCoolDownIntervalNs(TimeUnit.SECONDS.toNanos(1))
400-
.catchupEndpoint(hostname(index) + ":0")
409+
.catchupEndpoint(hostname(index) + ":" + catchupEndpointPort)
401410
.archiveContext(new AeronArchive.Context()
402-
.aeronDirectoryName(aeronDirName)
403-
.controlRequestChannel(context.archiveContext.localControlChannel())
404-
.controlRequestStreamId(context.archiveContext.localControlStreamId())
405-
.controlResponseChannel("aeron:ipc?alias=backup-archive-local-resp")
406-
.controlResponseStreamId(9090909 + index))
411+
.aeronDirectoryName(aeronDirName)
412+
.controlRequestChannel(context.archiveContext.localControlChannel())
413+
.controlRequestStreamId(context.archiveContext.localControlStreamId())
414+
.controlResponseChannel("aeron:ipc?alias=backup-archive-local-resp")
415+
.controlResponseStreamId(9090909 + index))
407416
.clusterArchiveContext(new AeronArchive.Context()
408-
.aeronDirectoryName(aeronDirName)
409-
.controlRequestChannel(context.archiveContext.controlChannel())
410-
.controlResponseChannel(archiveControlResponseChannel(index)))
417+
.aeronDirectoryName(aeronDirName)
418+
.controlRequestChannel(context.archiveContext.controlChannel())
419+
.controlResponseChannel(archiveControlResponseChannel(index)))
411420
.clusterDir(new File(baseDirName, "cluster-backup"))
412421
.credentialsSupplier(credentialsSupplier)
413422
.sourceType(sourceType)

0 commit comments

Comments
 (0)