Add an overload to Kafka extensions for registering multiple instances

[Alirexaa]

What this PR does / why we need it:
This pull request introduces a new overload for the AddKafka method in the KafkaHealthCheckBuilderExtensions class, which allows configuring Kafka health checks using an optionFactory. Additionally, the corresponding method signature is added to the approved API file.

New Overload for Kafka Health Check Configuration:

• src/HealthChecks.Kafka/DependencyInjection/KafkaHealthCheckBuilderExtensions.cs: Added a new overload of the AddKafka method that accepts an optionFactory to configure Kafka health checks. This method includes parameters for the health check name, failure status, tags, and timeout.

Update to Approved API:

• test/HealthChecks.Kafka.Tests/HealthChecks.Kafka.approved.txt: Added the new overload of the AddKafka method to the approved API file to reflect the changes in the method signature.
  Which issue(s) this PR fixes:

Please reference the issue this PR will close: #2298

Special notes for your reviewer:

Does this PR introduce a user-facing change?:

Please make sure you've completed the relevant tasks for this PR, out of the following list:

• [] Code compiles correctly
• Created/updated tests
• Unit tests passing
• End-to-end tests passing
• Extended the documentation
• Provided sample for the feature

[adamsitnik]

I would prefer to wait for feedback from Aspire if adding this particular overload is going to solve the problem. If so, we are going to need some new tests (just a few).

Also, I wonder if we should instead add an overload that requires a IProducer<K, V> factory .

Something like this:

public static IHealthChecksBuilder AddKafka<TKey, TValue>(
    this IHealthChecksBuilder builder,
    Func<ServiceProvider, IProducer<TKey, TValue>>? providerFactory = default,
    Func<Message<TKey, TValue>> messageBuilder,
    string topic = DEFAULT_TOPIC,
    string? name = default,
    HealthStatus? failureStatus = default,
    IEnumerable<string>? tags = default,
    TimeSpan? timeout = default)
{
    return builder.Add(new HealthCheckRegistration(
        name ?? NAME,
        sp => new KafkaHealthCheck<TKey, TValue>(providerFactory?.Invoke(sp) ?? sp.GetRequiredService<IProducer<TKey, TValue>>(), messageBuilder, topic),
        failureStatus,
        tags,
        timeout));
}

public class KafkaHealthCheck<TKey, TValue> : IHealthCheck, IDisposable
{
    private readonly IProducer<TKey, TValue> _producer;
    private readonly Func<Message<TKey, TValue>> _messageBuilder;
    private readonly string _topic;

    public KafkaHealthCheck(IProducer<TKey, TValue> producer, Func<Message<TKey, TValue>> messageBuilder, string? topic)
    {
        _producer = producer;
        _messageBuilder = messageBuilder;
        _topic = topic ?? KafkaHealthCheckBuilderExtensions.DEFAULT_TOPIC;
    }

    /// <inheritdoc />
    public async Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default)
    {
        try
        {
            Message<TKey, TValue> message = _messageBuilder.Invoke();

            var result = await _producer.ProduceAsync(_topic, message, cancellationToken).ConfigureAwait(false);

            if (result.Status == PersistenceStatus.NotPersisted)
            {
                return new HealthCheckResult(context.Registration.FailureStatus, description: $"Message is not persisted or a failure is raised on health check for kafka.");
            }

            return HealthCheckResult.Healthy();
        }
        catch (Exception ex)
        {
            return new HealthCheckResult(context.Registration.FailureStatus, exception: ex);
        }
    }

    public void Dispose() => _producer.Dispose();
}

This would force the users to register and resolve everything on their own in the DI and we would not need to care about settings.

[adamsitnik]

@mitchdenny would adding this overload solve the problem you have faced in Aspire?

I can see that you have solved it by simply reusing the ctor of the health check and registering your own thing in the DI container:

https://github.com/dotnet/aspire/blob/422d777c8dc7df154d9eea268f0d28a35e673f1b/src/Components/Aspire.Confluent.Kafka/AspireKafkaProducerExtensions.cs#L139-L151

[adamsitnik]

nit: whitespace

Suggested change

	Func<IServiceProvider, KafkaHealthCheckOptions> optionFactory,
	Func<IServiceProvider, KafkaHealthCheckOptions> optionFactory,

[Alirexaa]

@adamsitnik, I prefer that overload too but this mean we should remove all other overloads and introduce breaking changes. Should we do this?

[adamsitnik]

@adamsitnik, I prefer that overload too but this mean we should remove all other overloads and introduce breaking changes. Should we do this?

This time I am not 100% convinced as the new approach would require the users to always specify TKey, TValue and the message builder. So it would not be super easy to move to the new version.

I was hoping to get feedback from @mitchdenny but it seems that he just started his winter holiday leave.

@eerhardt what are your thoughts on this? Should we introduce such a breaking change or perhaps just extend the existing ways with a new overload?

[eerhardt]

@eerhardt what are your thoughts on this? Should we introduce such a breaking change or perhaps just extend the existing ways with a new overload?

My default opinion is to not introduce a breaking change if you can help it. If anything, we could create a new IHealthCheck class, if necessary.

[eerhardt]

I think this callback gets invoked every time the health is checked. So the optionFactory would be invoked each time as well.

[Alirexaa]

Correct.

[Alirexaa]

@eerhardt what are your thoughts on this? Should we introduce such a breaking change or perhaps just extend the existing ways with a new overload?

My default opinion is to not introduce a breaking change if you can help it. If anything, we could create a new IHealthCheck class, if necessary.

I thought about new class and it seems creating a new class is better approach. I think this new factory method could create memory leak.
Also with new class we can resolve singlton client from factory method and mark older class and methods as obsolated.

[Alirexaa]

@adamsitnik Should we add a new HealthCheck class?