Skip to content

Commit a8825c6

Browse files
lukebakkenpaulomorgado
lukebakken
and
paulomorgado
committed
Fix two flaky tests
These two tests always fail at the same time, too 🤔 * Re-try when using HTTP API to close a connection and an error happens * Introduce `ShortSpan` wait span. * Ensure that message is actually routed and confirmed before checking. * Limit client provided name to 3000 characters * Add callback exception handlers to flaky tests * Check for null consumer tag based on CI failures. * No need for random data, just different data. Update projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcherBase.cs Co-authored-by: Paulo Morgado <[email protected]>
1 parent 5da0720 commit a8825c6

File tree

13 files changed

+234
-114
lines changed

13 files changed

+234
-114
lines changed

‎projects/RabbitMQ.Client/client/api/InternalConstants.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,6 @@ internal static class InternalConstants
5050
/// This is not configurable, but was discovered while working on this issue:
5151
/// https://github.com/rabbitmq/rabbitmq-dotnet-client/issues/980
5252
/// </summary>
53-
internal const int DefaultRabbitMqMaxClientProvideNameLength = 3652;
53+
internal const int DefaultRabbitMqMaxClientProvideNameLength = 3000;
5454
}
5555
}

‎projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -279,7 +279,7 @@ public async Task<string> BasicConsumeAsync(string queue, bool autoAck, string c
279279
{
280280
string resultConsumerTag = await InnerChannel.BasicConsumeAsync(queue, autoAck, consumerTag, noLocal,
281281
exclusive, arguments, consumer, cancellationToken)
282-
.ConfigureAwait(false);
282+
.ConfigureAwait(false) ?? throw new InvalidOperationException("basic.consume returned null consumer tag");
283283
var rc = new RecordedConsumer(channel: this, consumer: consumer, consumerTag: resultConsumerTag,
284284
queue: queue, autoAck: autoAck, exclusive: exclusive, arguments: arguments);
285285
await _connection.RecordConsumerAsync(rc, recordedEntitiesSemaphoreHeld: false)

‎projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recording.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
// Copyright (c) 2007-2020 VMware, Inc. All rights reserved.
3030
//---------------------------------------------------------------------------
3131

32+
using System;
3233
using System.Collections.Generic;
3334
using System.Linq;
3435
using System.Threading;
@@ -400,6 +401,11 @@ await _recordedEntitiesSemaphore.WaitAsync()
400401

401402
private void DoDeleteRecordedConsumer(string consumerTag)
402403
{
404+
if (consumerTag is null)
405+
{
406+
throw new ArgumentNullException(nameof(consumerTag));
407+
}
408+
403409
if (_recordedConsumers.Remove(consumerTag, out RecordedConsumer recordedConsumer))
404410
{
405411
DeleteAutoDeleteQueue(recordedConsumer.Queue);

‎projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcherBase.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ namespace RabbitMQ.Client.ConsumerDispatching
1010
internal abstract class ConsumerDispatcherBase
1111
{
1212
private static readonly FallbackConsumer s_fallbackConsumer = new FallbackConsumer();
13-
private readonly IDictionary<string, IBasicConsumer> _consumers = new ConcurrentDictionary<string, IBasicConsumer>();
13+
private readonly ConcurrentDictionary<string, IBasicConsumer> _consumers = new ConcurrentDictionary<string, IBasicConsumer>();
1414

1515
public IBasicConsumer? DefaultConsumer { get; set; }
1616

‎projects/Test/Common/IntegrationFixture.cs‎

Lines changed: 32 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@ public abstract class IntegrationFixture : IAsyncLifetime
7575
protected readonly ushort _consumerDispatchConcurrency = 1;
7676
protected readonly bool _openChannel = true;
7777

78+
public static readonly TimeSpan ShortSpan;
7879
public static readonly TimeSpan WaitSpan;
7980
public static readonly TimeSpan LongWaitSpan;
8081
public static readonly TimeSpan RecoveryInterval = TimeSpan.FromSeconds(2);
@@ -95,12 +96,14 @@ static IntegrationFixture()
9596

9697
if (s_isRunningInCI)
9798
{
99+
ShortSpan = TimeSpan.FromSeconds(20);
98100
WaitSpan = TimeSpan.FromSeconds(60);
99101
LongWaitSpan = TimeSpan.FromSeconds(120);
100102
RequestedConnectionTimeout = TimeSpan.FromSeconds(4);
101103
}
102104
else
103105
{
106+
ShortSpan = TimeSpan.FromSeconds(10);
104107
WaitSpan = TimeSpan.FromSeconds(30);
105108
LongWaitSpan = TimeSpan.FromSeconds(60);
106109
}
@@ -160,9 +163,8 @@ public virtual async Task InitializeAsync()
160163
if (IsVerbose)
161164
{
162165
AddCallbackShutdownHandlers();
166+
AddCallbackExceptionHandlers();
163167
}
164-
165-
AddCallbackExceptionHandlers();
166168
}
167169

168170
if (_connFactory.AutomaticRecoveryEnabled)
@@ -221,59 +223,55 @@ protected virtual void DisposeAssertions()
221223

222224
protected void AddCallbackExceptionHandlers()
223225
{
224-
if (_conn != null)
226+
AddCallbackExceptionHandlers(_conn, _channel);
227+
}
228+
229+
protected void AddCallbackExceptionHandlers(IConnection conn, IChannel channel)
230+
{
231+
if (conn != null)
225232
{
226-
_conn.ConnectionRecoveryError += (s, ea) =>
233+
conn.ConnectionRecoveryError += (s, ea) =>
227234
{
228235
_connectionRecoveryException = ea.Exception;
229236

230-
if (IsVerbose)
237+
try
238+
{
239+
_output.WriteLine($"{0} connection recovery exception: {1}",
240+
_testDisplayName, _connectionRecoveryException);
241+
}
242+
catch (InvalidOperationException)
231243
{
232-
try
233-
{
234-
_output.WriteLine($"{0} connection recovery exception: {1}",
235-
_testDisplayName, _connectionRecoveryException);
236-
}
237-
catch (InvalidOperationException)
238-
{
239-
}
240244
}
241245
};
242246

243-
_conn.CallbackException += (o, ea) =>
247+
conn.CallbackException += (o, ea) =>
244248
{
245249
_connectionCallbackException = ea.Exception;
246250

247-
if (IsVerbose)
251+
try
252+
{
253+
_output.WriteLine("{0} connection callback exception: {1}",
254+
_testDisplayName, _connectionCallbackException);
255+
}
256+
catch (InvalidOperationException)
248257
{
249-
try
250-
{
251-
_output.WriteLine("{0} connection callback exception: {1}",
252-
_testDisplayName, _connectionCallbackException);
253-
}
254-
catch (InvalidOperationException)
255-
{
256-
}
257258
}
258259
};
259260
}
260261

261-
if (_channel != null)
262+
if (channel != null)
262263
{
263-
_channel.CallbackException += (o, ea) =>
264+
channel.CallbackException += (o, ea) =>
264265
{
265266
_channelCallbackException = ea.Exception;
266267

267-
if (IsVerbose)
268+
try
269+
{
270+
_output.WriteLine("{0} channel callback exception: {1}",
271+
_testDisplayName, _channelCallbackException);
272+
}
273+
catch (InvalidOperationException)
268274
{
269-
try
270-
{
271-
_output.WriteLine("{0} channel callback exception: {1}",
272-
_testDisplayName, _channelCallbackException);
273-
}
274-
catch (InvalidOperationException)
275-
{
276-
}
277275
}
278276
};
279277
}
@@ -491,11 +489,6 @@ protected static void AssertPreconditionFailed(ShutdownEventArgs args)
491489
AssertShutdownError(args, Constants.PreconditionFailed);
492490
}
493491

494-
protected static Task AssertRanToCompletion(params Task[] tasks)
495-
{
496-
return DoAssertRanToCompletion(tasks);
497-
}
498-
499492
protected static Task AssertRanToCompletion(IEnumerable<Task> tasks)
500493
{
501494
return DoAssertRanToCompletion(tasks);

‎projects/Test/Common/Util.cs‎

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -67,29 +67,32 @@ public static async Task CloseConnectionAsync(IConnection conn)
6767
connectionToClose = connections.Where(c0 =>
6868
string.Equals((string)c0.ClientProperties["connection_name"], conn.ClientProvidedName,
6969
StringComparison.InvariantCultureIgnoreCase)).FirstOrDefault();
70-
71-
if (connectionToClose == null)
72-
{
73-
tries++;
74-
}
75-
else
76-
{
77-
break;
78-
}
7970
}
8071
catch (ArgumentNullException)
8172
{
8273
// Sometimes we see this in GitHub CI
8374
tries++;
75+
continue;
8476
}
85-
} while (tries <= 30);
77+
78+
if (connectionToClose != null)
79+
{
80+
try
81+
{
82+
await s_managementClient.CloseConnectionAsync(connectionToClose);
83+
return;
84+
}
85+
catch (UnexpectedHttpStatusCodeException)
86+
{
87+
tries++;
88+
}
89+
}
90+
} while (tries <= 10);
8691

8792
if (connectionToClose == null)
8893
{
8994
throw new InvalidOperationException($"Could not delete connection: '{conn.ClientProvidedName}'");
9095
}
91-
92-
await s_managementClient.CloseConnectionAsync(connectionToClose);
9396
}
9497
}
9598
}

‎projects/Test/Integration/ConnectionRecovery/TestExchangeRecovery.cs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,8 @@ public async Task TestExchangeRecoveryTest()
5555
[Fact]
5656
public async Task TestExchangeToExchangeBindingRecovery()
5757
{
58+
await _channel.ConfirmSelectAsync();
59+
5860
string q = (await _channel.QueueDeclareAsync("", false, false, false)).QueueName;
5961

6062
string ex_source = GenerateExchangeName();
@@ -70,7 +72,8 @@ public async Task TestExchangeToExchangeBindingRecovery()
7072
{
7173
await CloseAndWaitForRecoveryAsync();
7274
Assert.True(_channel.IsOpen);
73-
await _channel.BasicPublishAsync(ex_source, "", _encoding.GetBytes("msg"));
75+
await _channel.BasicPublishAsync(ex_source, "", _encoding.GetBytes("msg"), mandatory: true);
76+
await _channel.WaitForConfirmsOrDieAsync();
7477
await AssertMessageCountAsync(q, 1);
7578
}
7679
finally

0 commit comments

Comments
 (0)