microsoft / microsoft/aspire

Can't configure Kafka health check timeouts

Open
#5,890 3 comments 0 reactions 0 assignees View on GitHub
area-integrations kafka
Dominant language
C#
Stars
6.3k
Forks
991
Avg merge
2d 15h
Merged PRs (30d)
196

Description

### Is there an existing issue for this?

- [X] I have searched the existing issues

### Is your feature request related to a problem? Please describe the problem.

I'm using the Kafka integration with Azure Event Hubs, which is working properly. However, I get a lot of connection errors as part of the default health checks enabled by the Kafka integration:

```
%3|1727184480.461|ERROR|rdkafka#producer-159| [thrd:sasl_ssl://eventstreamersfa5ktnjiw4.servicebus.windows.net:9093]: 1/1 brokers are down
%5|1727184481.949|REQTMOUT|rdkafka#producer-274| [thrd:sasl_ssl://eventstreamersfa5ktnjiw4.servicebus.windows.net:9093]: sasl_ssl://eventstreamersfa5ktnjiw4.servicebus.windows.net:9093/0: Timed out MetadataRequest in flight (after 1000ms, timeout #0)
%3|1727184481.949|FAIL|rdkafka#producer-274| [thrd:sasl_ssl://eventstreamersfa5ktnjiw4.servicebus.windows.net:9093]: sasl_ssl://eventstreamersfa5ktnjiw4.servicebus.windows.net:9093/0: 1 request(s) timed out: disconnect (average rtt 5.565ms) (after 300970ms in state UP)
%3|1727184481.956|ERROR|rdkafka#producer-4| [thrd:app]: rdkafka#producer-4: sasl_ssl://eventstreamersfa5ktnjiw4.servicebus.windows.net:9093/0: 1 request(s) timed out: disconnect (average rtt 2.908ms) (after 299975ms in state UP)
```

I already configured the Kafka integration with higher timeout values, but those won't apply to health checks since they seem to be hardcoded in [AddKafkaProducerInternal](https://github.com/dotnet/aspire/blob/668ade88500c28bf139f400dec7699e6f356b5aa/src/Components/Aspire.Confluent.Kafka/AspireKafkaProducerExtensions.cs#L103):

```csharp
builder.TryAddHealthCheck(new HealthCheckRegistration(healthCheckName,
sp =>
{
var connectionFactory = serviceKey is null
? sp.GetRequiredService>()
: sp.GetRequiredKeyedService>(serviceKey);

var options = new KafkaHealthCheckOptions();
options.Configuration = new ProducerConfig(connectionFactory.Config.ToDictionary());
options.Configuration.SocketTimeoutMs = 1000;
options.Configuration.MessageTimeoutMs = 1000;
options.Configuration.StatisticsIntervalMs = 0;
return new KafkaHealthCheck(options);
},
failureStatus: default,
tags: default));
```

### Describe the solution you'd like

Can those hardcoded values be removed? They should just use whatever is already set in the provided producer config.

### Additional context

_No response_

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.