diff --git a/src/Cassandra.IntegrationTests/Core/HeartbeatTests.cs b/src/Cassandra.IntegrationTests/Core/HeartbeatTests.cs index 1c18b38fc..3cf682937 100644 --- a/src/Cassandra.IntegrationTests/Core/HeartbeatTests.cs +++ b/src/Cassandra.IntegrationTests/Core/HeartbeatTests.cs @@ -19,6 +19,7 @@ using Cassandra.IntegrationTests.SimulacronAPI.Models.Logs; using Cassandra.IntegrationTests.TestBase; using Cassandra.IntegrationTests.TestClusterManagement.Simulacron; +using Cassandra.SessionManagement; using Cassandra.Tests; using NUnit.Framework; @@ -71,10 +72,55 @@ await TestHelper.RetryAssertAsync( logs = await _testCluster.GetNodes().First() .GetQueriesAsync(null, OptionsQueryType).ConfigureAwait(false); Assert.That(logs.Count, Is.GreaterThan(initialCount)); - }, + }, 500, 20).ConfigureAwait(false); } } + + [Test] + public async Task Connection_Should_Be_Closed_When_Heartbeat_Times_Out_With_Socket_Still_Open() + { + var builder = ClusterBuilder() + .WithPoolingOptions(PoolingOptions.Create() + .SetHeartBeatInterval(2000) + .SetCoreConnectionsPerHost(HostDistance.Local, 1) + .SetMaxConnectionsPerHost(HostDistance.Local, 1)) + .WithSocketOptions(new SocketOptions() + .SetReadTimeoutMillis(2000) + .SetDefunctReadTimeoutThreshold(int.MaxValue)) + .AddContactPoint(_testCluster.InitialContactPoint); + + using (var cluster = builder.Build()) + { + var session = await cluster.ConnectAsync().ConfigureAwait(false); + await session.ExecuteAsync(new SimpleStatement(HeartbeatTests.Query)).ConfigureAwait(false); + + var pool = ((IInternalSession)session).GetPools().Single().Value; + var connection = pool.ConnectionsSnapshot.Single(); + Assert.IsFalse(connection.IsDisposed); + + // Simulacron keeps the TCP connection ESTABLISHED and just stops reading, so the + // heartbeat fails with OperationTimedOutException rather than SocketException. No + // application requests are sent after this point, so only the heartbeat itself can + // detect the unresponsive host and close the connection. + await _testCluster.PauseReadsAsync().ConfigureAwait(false); + try + { + await TestHelper.RetryAssertAsync( + () => + { + Assert.IsTrue(connection.IsDisposed); + return Task.CompletedTask; + }, + 500, + 20).ConfigureAwait(false); + } + finally + { + await _testCluster.ResumeReadsAsync().ConfigureAwait(false); + } + } + } } } \ No newline at end of file diff --git a/src/Cassandra/Connections/Connection.cs b/src/Cassandra/Connections/Connection.cs index 720268ef1..bb4d9e6c5 100644 --- a/src/Cassandra/Connections/Connection.cs +++ b/src/Cassandra/Connections/Connection.cs @@ -438,7 +438,7 @@ private void IdleTimeoutHandler(object state) return TaskHelper.Completed; } Connection.Logger.Warning("Received heartbeat request exception " + error.Exception.ToString()); - if (error.Exception is SocketException) + if (error.Exception is SocketException || error.Exception is OperationTimedOutException) { OnIdleRequestException?.Invoke(error.Exception); }