How to stop kafkaflow gracefully

[razondpro]

Hi,

When the brokers are down, i want to stop the application gracefully. The problem is that kafkaflow keeps running even after
trying to stop the application forcefully (ctr+c).

The errors:

%3|1696234620.026|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) %3|1696234620.026|ERROR|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: 1/1 brokers are down %3|1696234620.026|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) %3|1696234622.030|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
It keeps trying to connect to the brokers. How to can I Stop it?
it continues until getting an expetion:

Unhandled exception. Confluent.Kafka.KafkaException: Failed while waiting for controller: Local: Timed out at KafkaFlow.Clusters.ClusterManager.CreateIfNotExistsAsync(IEnumerable1 configurations)
at KafkaFlow.KafkaBus.CreateMissingClusterTopics(ClusterConfiguration cluster)
at KafkaFlow.KafkaBus.StartAsync(CancellationToken stopCancellationToken)
at Microsoft.Extensions.Hosting.Internal.Host.StartAsync(CancellationToken cancellationToken)
at Microsoft.Extensions.Hosting.HostingAbstractionsHostExtensions.RunAsync(IHost host, CancellationToken token)
at Microsoft.Extensions.Hosting.HostingAbstractionsHostExtensions.RunAsync(IHost host, CancellationToken token)`

my code:
` public void Install(IServiceCollection services, IConfiguration configuration)
{
var kafkaOptionss = configuration.GetSection(BusOptionsSetup.ConfigurationSectionName).Get()!;
const string usersTopic = "users";
const string consumerName = "users-consumer";

        services.AddSingleton<KafkaProducer>();
        services.ConfigureOptions<BusOptionsSetup>();

        services.AddKafkaFlowHostedService(configuration =>
        {
            configuration.UseLogHandler<CustomLogHandler>();
            configuration.AddCluster(cluster =>
            {
                cluster.WithBrokers(kafkaOptionss.BootstrapServers);
                cluster.WithSchemaRegistry(config => config.Url = kafkaOptionss.SchemaRegistryUrl);
                cluster.CreateTopicIfNotExists(usersTopic, 1, 1);

                cluster.AddConsumer(consumer =>
                {
                    consumer.Topics(kafkaOptionss.Topics);
                    consumer.WithName(consumerName);
                    consumer.WithGroupId(kafkaOptionss.GroupId);
                    consumer.WithBufferSize(100);
                    consumer.WithWorkersCount(3);
                    consumer.WithAutoOffsetReset(AutoOffsetReset.Earliest);
                    consumer.WithManualStoreOffsets();
                    consumer.AddMiddlewares(middlewares => middlewares
                        .AddSchemaRegistryAvroSerializer()
                        .AddTypedHandlers(handlers => handlers
                            // Transient needed because of mediatr(is transient by default)
                            .WithHandlerLifetime(InstanceLifetime.Transient)
                            .AddHandler<UserCreatedConsumerEventHandler>()
                        )
                    );
                });

                cluster.AddProducer<KafkaProducer>(producer =>
                {
                    producer.DefaultTopic(usersTopic);
                    producer.WithAcks(Acks.All);
                    producer.AddMiddlewares(middlewares =>
                    {
                        middlewares.AddSchemaRegistryAvroSerializer(
                            new AvroSerializerConfig
                            {
                                SubjectNameStrategy = SubjectNameStrategy.TopicRecord,
                                AutoRegisterSchemas = true
                            }
                        );
                    });
                });

            });
        });
    }

`

Thank you.