Skip to content

Commit c9a549e

Browse files
niemyjskiniemyjski
authored
fix: adapt Kafka MessageBus to two-phase dispose lifecycle (#54)
* fix: adapt Kafka MessageBus to two-phase dispose lifecycle - Add CleanupAsync override for listener task, producer flush/dispose - Handle OperationCanceledException in OnMessageAsync to skip offset commit - Guard offset commit with IsDisposed check - Remove ad-hoc Dispose override - Depends on FoundatioFx/Foundatio#492 * Update CanDisposeWithNoSubscribersOrPublishers to async * Order MessageBus lifecycle test overrides alphabetically * Address PR feedback: fix offset commit after OCE, add disposal filter, harden CleanupAsync * Update Foundatio dependencies to 13.0.0-beta5.2 * Fix DeleteTopicAsync disposal guard and cancellation support * Address PR feedback: restore removed comment blocks, fix single-line catch formatting in test * Commit offset when no subscribers and add comment to OperationCanceledException catch * Add comments explaining Kafka offset acknowledgment logic * Fix: do not acknowledge messages when no subscribers to preserve Kafka redelivery
1 parent d1bfd38 commit c9a549e

5 files changed

Lines changed: 104 additions & 36 deletions

File tree

‎src/Foundatio.Kafka/Foundatio.Kafka.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
<ItemGroup>
33
<PackageReference Include="Confluent.Kafka" Version="2.14.0" />
44

5-
<PackageReference Include="Foundatio" Version="13.0.0-beta5" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
5+
<PackageReference Include="Foundatio" Version="13.0.0-beta5.2" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
66
<ProjectReference Include="..\..\..\Foundatio\src\Foundatio\Foundatio.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
77
</ItemGroup>
88
</Project>

‎src/Foundatio.Kafka/Messaging/KafkaMessageBus.cs‎

Lines changed: 37 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ namespace Foundatio.Messaging;
1414

1515
public interface IKafkaMessageBus : IMessageBus
1616
{
17-
Task DeleteTopicAsync();
17+
Task DeleteTopicAsync(CancellationToken cancellationToken = default);
1818
}
1919

2020
public class KafkaMessageBus : MessageBusBase<KafkaMessageBusOptions>, IKafkaMessageBus
@@ -26,7 +26,6 @@ public class KafkaMessageBus : MessageBusBase<KafkaMessageBusOptions>, IKafkaMes
2626
private readonly AsyncLock _lock = new();
2727
private readonly AsyncManualResetEvent _consumerReady = new();
2828
private bool _topicCreated;
29-
private bool _isDisposed;
3029

3130
public KafkaMessageBus(KafkaMessageBusOptions options) : base(options)
3231
{
@@ -96,6 +95,9 @@ protected override async Task PublishImplAsync(string messageType, object messag
9695

9796
private async Task OnMessageAsync(IConsumer<string, byte[]> consumer, ConsumeResult<string, byte[]> consumeResult)
9897
{
98+
// No subscribers registered — skip processing but do NOT commit the offset.
99+
// Kafka retains uncommitted messages so they are redelivered once a subscriber
100+
// reconnects, which is the expected persistence-guarantee behavior.
99101
if (_subscribers.IsEmpty)
100102
return;
101103

@@ -105,6 +107,7 @@ private async Task OnMessageAsync(IConsumer<string, byte[]> consumer, ConsumeRes
105107

106108
_logger.LogTrace("OnMessage(TopicPartitionOffset={TopicPartitionOffset} GroupId={GroupId})", consumeResult.TopicPartitionOffset, _consumerConfig.GroupId);
107109

110+
bool shouldAcknowledge = true;
108111
try
109112
{
110113
string? messageType = _options.ResolveMessageType?.Invoke(consumeResult);
@@ -124,6 +127,11 @@ private async Task OnMessageAsync(IConsumer<string, byte[]> consumer, ConsumeRes
124127
var message = ConvertToMessage(messageType, consumeResult.Message);
125128
await SendMessageToSubscribersAsync(message).AnyContext();
126129
}
130+
catch (OperationCanceledException) when (IsDisposed)
131+
{
132+
// Bus is disposing — skip offset commit so the message is redelivered to other consumers
133+
shouldAcknowledge = false;
134+
}
127135
catch (MessageBusException)
128136
{
129137
// Subscriber handler failures are application-level errors. The base class
@@ -138,7 +146,9 @@ private async Task OnMessageAsync(IConsumer<string, byte[]> consumer, ConsumeRes
138146
}
139147
finally
140148
{
141-
if (!_subscribers.IsEmpty)
149+
// Commit the offset unless the bus is disposing, in which case we skip so
150+
// the message is redelivered to other consumers in the group.
151+
if (shouldAcknowledge && !IsDisposed)
142152
AcknowledgeMessage(consumer, consumeResult);
143153
}
144154
}
@@ -295,23 +305,33 @@ private void EnsureListening()
295305
}, DisposedCancellationToken);
296306
}
297307

298-
public override void Dispose()
308+
protected override async Task CleanupAsync()
299309
{
300-
if (_isDisposed)
310+
if (_listeningTask is not null)
301311
{
302-
_logger.LogTrace("MessageBus {MessageBusId} dispose was already called", MessageBusId);
303-
return;
312+
try
313+
{
314+
await _listeningTask.WaitAsync(TimeSpan.FromSeconds(15)).AnyContext();
315+
}
316+
catch (OperationCanceledException ex)
317+
{
318+
_logger.LogTrace(ex, "Listening task cancelled during cleanup");
319+
}
320+
catch (TimeoutException)
321+
{
322+
_logger.LogWarning("Listening task did not complete within timeout during dispose");
323+
}
324+
catch (Exception ex)
325+
{
326+
_logger.LogWarning(ex, "Error waiting for listening task to complete during cleanup");
327+
}
304328
}
305329

306-
_isDisposed = true;
307-
_logger.LogTrace("MessageBus {MessageBusId} dispose", MessageBusId);
308-
309330
int? queueSize = _producer?.Flush(TimeSpan.FromSeconds(15));
310331
if (queueSize > 0)
311332
_logger.LogTrace("Flushed producer {queueSize}", queueSize);
312333

313334
_producer?.Dispose();
314-
base.Dispose();
315335
}
316336

317337
private async Task EnsureTopicCreatedAsync()
@@ -384,9 +404,13 @@ private async Task EnsureTopicCreatedAsync()
384404
}
385405
}
386406

387-
async Task IKafkaMessageBus.DeleteTopicAsync()
407+
async Task IKafkaMessageBus.DeleteTopicAsync(CancellationToken cancellationToken)
388408
{
389-
using var topicLock = await _lock.LockAsync(DisposedCancellationToken).AnyContext();
409+
ObjectDisposedException.ThrowIf(IsDisposed, this);
410+
411+
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
412+
cancellationToken, DisposedCancellationToken);
413+
using var topicLock = await _lock.LockAsync(linkedCts.Token).AnyContext();
390414
using var adminClient = new AdminClientBuilder(_adminClientConfig)
391415
.SetLogHandler(LogHandler)
392416
.SetStatisticsHandler(LogStatisticsHandler)

‎tests/Directory.Build.props‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.5" />
1313
<PackageReference Include="GitHubActionsTestLogger" Version="3.0.3" PrivateAssets="All" />
1414

15-
<PackageReference Include="Foundatio.TestHarness" Version="13.0.0-beta5" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
15+
<PackageReference Include="Foundatio.TestHarness" Version="13.0.0-beta5.2" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
1616
<ProjectReference Include="..\..\..\Foundatio\src\Foundatio.TestHarness\Foundatio.TestHarness.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
1717
</ItemGroup>
1818
</Project>

‎tests/Foundatio.Kafka.Tests/Messaging/KafkaMessageBusTestBase.cs‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,15 @@ public KafkaMessageBusTestBase(ITestOutputHelper output) : base(output)
3838
protected override async Task CleanupMessageBusAsync(IMessageBus messageBus)
3939
{
4040
if (EnableTopicDeletion && messageBus is IKafkaMessageBus kafkaMessageBus)
41-
await kafkaMessageBus.DeleteTopicAsync();
41+
{
42+
try
43+
{
44+
await kafkaMessageBus.DeleteTopicAsync();
45+
}
46+
catch (ObjectDisposedException)
47+
{
48+
}
49+
}
4250

4351
await base.CleanupMessageBusAsync(messageBus);
4452
}

‎tests/Foundatio.Kafka.Tests/Messaging/KafkaMessageBusTests.cs‎

Lines changed: 56 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -45,21 +45,9 @@ public override Task CanSendDelayedMessageAsync()
4545
}
4646

4747
[Fact]
48-
public override Task PublishAsync_WithCancellation_ThrowsOperationCanceledExceptionAsync()
49-
{
50-
return base.PublishAsync_WithCancellation_ThrowsOperationCanceledExceptionAsync();
51-
}
52-
53-
[Fact]
54-
public override Task PublishAsync_WithDelayedMessageAndDisposeBeforeDelivery_DiscardsMessageAsync()
55-
{
56-
return base.PublishAsync_WithDelayedMessageAndDisposeBeforeDelivery_DiscardsMessageAsync();
57-
}
58-
59-
[Fact]
60-
public override Task SubscribeAsync_WithCancellation_ThrowsOperationCanceledExceptionAsync()
48+
public override Task CanSendMappedMessageAsync()
6149
{
62-
return base.SubscribeAsync_WithCancellation_ThrowsOperationCanceledExceptionAsync();
50+
return base.CanSendMappedMessageAsync();
6351
}
6452

6553
[Fact]
@@ -117,9 +105,9 @@ public override Task WontKeepMessagesWithNoSubscribersAsync()
117105
}
118106

119107
[Fact]
120-
public override void CanDisposeWithNoSubscribersOrPublishers()
108+
public override Task CanDisposeWithNoSubscribersOrPublishersAsync()
121109
{
122-
base.CanDisposeWithNoSubscribersOrPublishers();
110+
return base.CanDisposeWithNoSubscribersOrPublishersAsync();
123111
}
124112

125113
[Fact]
@@ -129,9 +117,39 @@ public override Task CanHandlePoisonedMessageAsync()
129117
}
130118

131119
[Fact]
132-
public override Task SubscribeAsync_WithValidThenPoisonedMessage_DeliversOnlyValidMessageAsync()
120+
public override Task DisposeAsync_CalledMultipleTimes_IsIdempotentAsync()
133121
{
134-
return base.SubscribeAsync_WithValidThenPoisonedMessage_DeliversOnlyValidMessageAsync();
122+
return base.DisposeAsync_CalledMultipleTimes_IsIdempotentAsync();
123+
}
124+
125+
[Fact]
126+
public override Task DisposeAsync_WhilePublishing_CompletesWithoutDeadlockAsync()
127+
{
128+
return base.DisposeAsync_WhilePublishing_CompletesWithoutDeadlockAsync();
129+
}
130+
131+
[Fact]
132+
public override Task DisposeAsync_WithNoSubscribersOrPublishers_CompletesWithoutExceptionAsync()
133+
{
134+
return base.DisposeAsync_WithNoSubscribersOrPublishers_CompletesWithoutExceptionAsync();
135+
}
136+
137+
[Fact]
138+
public override Task PublishAsync_AfterDispose_ThrowsMessageBusExceptionAsync()
139+
{
140+
return base.PublishAsync_AfterDispose_ThrowsMessageBusExceptionAsync();
141+
}
142+
143+
[Fact]
144+
public override Task PublishAsync_WithCancellation_ThrowsOperationCanceledExceptionAsync()
145+
{
146+
return base.PublishAsync_WithCancellation_ThrowsOperationCanceledExceptionAsync();
147+
}
148+
149+
[Fact]
150+
public override Task PublishAsync_WithDelayedMessageAndDisposeBeforeDelivery_DiscardsMessageAsync()
151+
{
152+
return base.PublishAsync_WithDelayedMessageAndDisposeBeforeDelivery_DiscardsMessageAsync();
135153
}
136154

137155
[Fact]
@@ -140,16 +158,34 @@ public override Task PublishAsync_WithSerializationFailure_ThrowsSerializerExcep
140158
return base.PublishAsync_WithSerializationFailure_ThrowsSerializerExceptionAsync();
141159
}
142160

161+
[Fact]
162+
public override Task SubscribeAsync_AfterDispose_ThrowsMessageBusExceptionAsync()
163+
{
164+
return base.SubscribeAsync_AfterDispose_ThrowsMessageBusExceptionAsync();
165+
}
166+
167+
[Fact]
168+
public override Task SubscribeAsync_CancelledToken_DoesNotTearDownInfrastructureAsync()
169+
{
170+
return base.SubscribeAsync_CancelledToken_DoesNotTearDownInfrastructureAsync();
171+
}
172+
173+
[Fact]
174+
public override Task SubscribeAsync_WithCancellation_ThrowsOperationCanceledExceptionAsync()
175+
{
176+
return base.SubscribeAsync_WithCancellation_ThrowsOperationCanceledExceptionAsync();
177+
}
178+
143179
[Fact]
144180
public override Task SubscribeAsync_WithDeserializationFailure_SkipsMessageAsync()
145181
{
146182
return base.SubscribeAsync_WithDeserializationFailure_SkipsMessageAsync();
147183
}
148184

149185
[Fact]
150-
public override Task CanSendMappedMessageAsync()
186+
public override Task SubscribeAsync_WithValidThenPoisonedMessage_DeliversOnlyValidMessageAsync()
151187
{
152-
return base.CanSendMappedMessageAsync();
188+
return base.SubscribeAsync_WithValidThenPoisonedMessage_DeliversOnlyValidMessageAsync();
153189
}
154190

155191
[Fact]

0 commit comments

Comments
 (0)