Skip to content

Commit 797d19e

Browse files
mjpt777vyazelenko
authored andcommitted
[Java] Log event for unknown session close in service container rather than throw and exception.
(cherry picked from commit 752cc9c)
1 parent 5f90f54 commit 797d19e

2 files changed

Lines changed: 45 additions & 17 deletions

File tree

aeron-cluster/src/main/java/io/aeron/cluster/service/ClusteredServiceAgent.java

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -478,26 +478,28 @@ void onSessionClose(
478478
{
479479
this.logPosition = logPosition;
480480
clusterTime = timestamp;
481-
final ContainerClientSession session = sessionByIdMap.remove(clusterSessionId);
482481

482+
final ContainerClientSession session = sessionByIdMap.remove(clusterSessionId);
483483
if (null == session)
484484
{
485-
throw new ClusterException(
486-
"unknown clusterSessionId=" + clusterSessionId + " for close reason=" + closeReason +
487-
" leadershipTermId=" + leadershipTermId + " logPosition=" + logPosition);
485+
ctx.countedErrorHandler().onError(new ClusterEvent(
486+
"unknown session close: clusterSessionId=" + clusterSessionId + " closeReason=" + closeReason +
487+
" leadershipTermId=" + leadershipTermId + " logPosition=" + logPosition));
488488
}
489-
490-
for (int i = 0, size = sessions.size(); i < size; i++)
489+
else
491490
{
492-
if (sessions.get(i).id() == clusterSessionId)
491+
for (int i = 0, size = sessions.size(); i < size; i++)
493492
{
494-
sessions.remove(i);
495-
break;
493+
if (sessions.get(i).id() == clusterSessionId)
494+
{
495+
sessions.remove(i);
496+
break;
497+
}
496498
}
497-
}
498499

499-
session.disconnect(ctx.countedErrorHandler());
500-
service.onSessionClose(session, timestamp, closeReason);
500+
session.disconnect(ctx.countedErrorHandler());
501+
service.onSessionClose(session, timestamp, closeReason);
502+
}
501503
}
502504

503505
void onServiceAction(

aeron-cluster/src/test/java/io/aeron/cluster/service/ClusteredServiceAgentTest.java

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,31 +15,33 @@
1515
*/
1616
package io.aeron.cluster.service;
1717

18-
import io.aeron.Aeron;
19-
import io.aeron.ConcurrentPublication;
20-
import io.aeron.Publication;
21-
import io.aeron.Subscription;
22-
import io.aeron.UnavailableCounterHandler;
18+
import io.aeron.*;
2319
import io.aeron.cluster.client.AeronCluster;
20+
import io.aeron.cluster.codecs.CloseReason;
2421
import io.aeron.driver.DutyCycleTracker;
2522
import io.aeron.logbuffer.BufferClaim;
2623
import io.aeron.test.CountersAnswer;
2724
import io.aeron.test.Tests;
2825
import org.agrona.DirectBuffer;
26+
import org.agrona.ErrorHandler;
2927
import org.agrona.MutableDirectBuffer;
3028
import org.agrona.concurrent.*;
3129
import org.agrona.concurrent.errors.DistinctErrorLog;
30+
import org.agrona.concurrent.errors.ErrorLogReader;
31+
import org.agrona.concurrent.status.AtomicCounter;
3232
import org.agrona.concurrent.status.CountersManager;
3333
import org.junit.jupiter.api.Test;
3434
import org.mockito.ArgumentCaptor;
3535

36+
import java.nio.ByteBuffer;
3637
import java.util.concurrent.TimeUnit;
3738

3839
import static io.aeron.Aeron.NULL_VALUE;
3940
import static io.aeron.AeronCounters.CLUSTER_COMMIT_POSITION_TYPE_ID;
4041
import static io.aeron.AeronCounters.CLUSTER_RECOVERY_STATE_TYPE_ID;
4142
import static org.junit.jupiter.api.Assertions.assertEquals;
4243
import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
44+
import static org.junit.jupiter.api.Assertions.assertTrue;
4345
import static org.mockito.Mockito.*;
4446

4547
class ClusteredServiceAgentTest
@@ -124,4 +126,28 @@ void shouldAbortClusteredServiceIfCommitPositionCounterIsClosed()
124126
nanoClock.advance(TimeUnit.MILLISECONDS.toNanos(2));
125127
assertThrowsExactly(ClusterTerminationException.class, clusteredServiceAgent::doWork);
126128
}
129+
130+
@Test
131+
void shouldLogErrorInsteadOfThrowingIfSessionIsNotFoundOnClose()
132+
{
133+
final Aeron aeron = mock(Aeron.class);
134+
final DistinctErrorLog distinctErrorLog = new DistinctErrorLog(
135+
new UnsafeBuffer(ByteBuffer.allocateDirect(16384)), new SystemEpochClock());
136+
final CountersManager countersManager = Tests.newCountersMananger(16 * 1024);
137+
final AtomicCounter errorCounter = countersManager.newCounter("test");
138+
final long originalErrorCount = errorCounter.get();
139+
140+
final ErrorHandler errorHandler = CommonContext.setupErrorHandler(null, distinctErrorLog);
141+
final CountedErrorHandler countedErrorHandler = new CountedErrorHandler(errorHandler, errorCounter);
142+
final ClusteredServiceContainer.Context ctx = new ClusteredServiceContainer.Context()
143+
.aeron(aeron)
144+
.idleStrategySupplier(() -> YieldingIdleStrategy.INSTANCE)
145+
.countedErrorHandler(countedErrorHandler);
146+
final ClusteredServiceAgent clusteredServiceAgent = new ClusteredServiceAgent(ctx);
147+
148+
clusteredServiceAgent.onSessionClose(99, 999, 9999, 99999, CloseReason.CLIENT_ACTION);
149+
150+
assertEquals(originalErrorCount + 1, errorCounter.get());
151+
assertTrue(ErrorLogReader.hasErrors(distinctErrorLog.buffer()));
152+
}
127153
}

0 commit comments

Comments
 (0)