Question: How to use Confluent Kafka .NET With/Without Polly To Implement A Retry and Wait Failure Policy For A Kafka AdminClient That Makes A Request To Create A Topic On The Broker?

Hi,
I am fairly new to Kafka and Polly. I am seeking advice with respect to how to implement failure resiliency when using the Admin Client with Kakfa Confluent .NET client. I am using the Admin Client to create a topic if it does not already exist during startup of a Blazor Server Web App.
Initially, I am trying to use polly to implement a simple wait and retry policy, listed below. I am expecting this to retry a create topic operation for a configurable number of attempts. Between each retry attempt there is a short configurable wait delay. If all retry attempts have been exhausted then a fatal error is signalled and the application gracefully exits.
Wait and Retry Policy
public static AsyncRetryPolicy BrokerWaitAndRetry(short retryCount, short waitSeconds, ILogger logger)
{
    var pauseBetweenFailures = TimeSpan.FromSeconds(waitSeconds);

    var retryPolicy = Policy
        .Handle<KafkaException>()
        .WaitAndRetryAsync(retryCount, i => pauseBetweenFailures, onRetry: (Exception e, TimeSpan time) =>
        {
            logger.LogInformation($"Retry attempt {time}");
        });
    }
    return retryPolicy;
}

Using the Wait and Retry Policy With An Admin Client Instance To Create A Topic
var polly = KakaFailurePolicies.BrokerWaitAndRetry(5, 7, _Logger);
await polly.ExecuteAsync(async () =>
{
  await client.CreateTopicsAsync(
    new TopicSpecification[] {
      new TopicSpecification {
        Name = _Config.Topic.Name,
        NumPartitions = _Config.Topic.PartitionCount,
        ReplicationFactor = _Config.Topic.ReplicationCount
      }
    },
    new CreateTopicsOptions
    {
        /** Set low request timeout for basic testing, otherwise would use a higher timeout of 60 seconds*/
        RequestTimeout = TimeSpan.FromSeconds(20)
    }
  );
});

Logs and Analysis
When I try and run this I can see from the logs below that a retry is attempted once. However, no subsequent retries are attempted. The logs highlight that an identical error has been detected and suppressed by the rdkafka library.
I think this is the reason that there are no subsequent retry attempts shown in the logs, i.e. the underlying rdkafka library poll thread is hiding subsequent local timeout errors and continues to try and connect to the broker. Subsequently the .NET client is unable to throw the exception since it has not been notified of the failure. This means that polly is stuck awaiting and running the second attempt?
info: Extensions.Hosting.AsyncInitialization.RootInitializer[0]
      Starting async initialization
info: Extensions.Hosting.AsyncInitialization.RootInitializer[0]
      Starting async initialization for WebApp.Kafka.Admin.KafkaAdminService
%3|1609337520.079|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 2ms in state CONNECT)
%3|1609337520.079|ERROR|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: 1/1 brokers are down
%3|1609337520.082|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 2ms in state CONNECT)
info: WebApp.Kafka.Admin.KafkaAdminService[0]
      Admin service trying to create Kafka Topic...
info: WebApp.Kafka.Admin.KafkaAdminService[0]
      Topic::eventbus, ReplicationCount::1, PartitionCount::3
info: WebApp.Kafka.Admin.KafkaAdminService[0]
      Bootstrap Servers::localhost:9092
%3|1609337521.081|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
%3|1609337521.081|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
%3|1609337522.083|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 1ms in state CONNECT)
%3|1609337522.083|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 1ms in state CONNECT)
%3|1609337524.089|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337524.089|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337525.090|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|1609337525.090|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|1609337527.097|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337527.097|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337528.096|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|1609337528.096|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|1609337530.106|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337530.106|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337532.115|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|1609337532.115|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|1609337533.119|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337533.119|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337535.128|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|1609337535.128|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|1609337536.128|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337536.128|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337537.129|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|1609337537.129|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|1609337540.133|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337540.133|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
Exception thrown: 'Confluent.Kafka.KafkaException' in System.Private.CoreLib.dll
info: WebApp.Kafka.Admin.KafkaAdminService[0]
      Retry attempt 00:00:07
%3|1609337542.135|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|1609337542.135|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|1609337544.140|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337544.140|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337545.140|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|1609337545.140|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|1609337546.140|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337546.140|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337547.332|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|1609337547.332|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|1609337548.149|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337548.149|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337549.148|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|1609337549.148|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|1609337550.148|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337550.148|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337551.152|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
%3|1609337551.152|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
%3|1609337555.171|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|1609337555.171|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|1609337556.171|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337556.171|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337557.177|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|1609337557.177|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|1609337558.176|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337558.176|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337560.180|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|1609337560.180|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|1609337561.183|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337561.183|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337562.186|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|1609337562.186|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|1609337563.186|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337563.186|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337564.191|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|1609337564.191|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|1609337567.200|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337567.200|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337568.204|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|1609337568.204|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|1609337569.208|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337569.208|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337570.212|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 1ms in state CONNECT)
%3|1609337570.212|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 1ms in state CONNECT)
%3|1609337573.222|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337573.222|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337574.225|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|1609337574.226|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|1609337575.230|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337575.230|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337576.229|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|1609337576.229|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|1609337578.234|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337578.234|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337579.238|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|1609337579.238|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|1609337580.241|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337580.241|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337581.245|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
%3|1609337581.245|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
%3|1609337583.254|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|1609337583.254|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|1609337584.259|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337584.259|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337585.264|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 1ms in state CONNECT)
%3|1609337585.264|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 1ms in state CONNECT)
%3|1609337586.267|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337586.267|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337587.267|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|1609337587.267|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|1609337588.267|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337588.267|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337589.268|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|1609337589.268|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|1609337591.278|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337591.278|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337592.282|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|1609337592.282|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|1609337593.286|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1609337593.286|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)

The error that is thrown by the Confluent Kafka .NET library is ErrorCode.Local_TimedOut during the first attempt. I think this corresponds to rdkafka:error local timeout??
Upon investigation I found the following AdminClient configuration properties relating to transient fault handling:

SocketTimeoutMs
SocketMaxFails
ReconnectBackoffMs
ReconnectBackoffMaxMs

This led me to try Confluent Kafka's built in transient fault handling:
var adminClientBuilder = new AdminClientBuilder(
/** https://github.com/confluentinc/confluent-kafka-dotnet/blob/2b54dd2ab07f6eaa886d88c57143838627b2512f/src/Confluent.Kafka/Config_gen.cs#L826 **/
    new AdminClientConfig()
    {
        BootstrapServers = _Config.Consumer.BootstrapServers,
        Debug = "all",
        SocketTimeoutMs = 20000, // reduced from 60seconds for testing purposes
        SocketMaxFails = 5, // is this synonymous to 5 retries?
    }
 );

After waiting for a short while during startup, I then started the Kafka broker. The debug logs below show that rdkafka thread detects that the main broker is up and running. However, the Create Topics admin operation is not being performed.
Debug logs
To reduce message size, included on pastebin.
Questions
Is it possible to configure rdkafka so that identical errors are not suppressed when trying to connect the AdminClient to the broker?
In the above example, once the rdkafka poll thread has eventually detected that the broker is up and running, why does the AdminClient create topics operation not complete?
How to use Confluent Kafka .NET with/without Polly to implement a retry and wait failure policy for a Kafka AdminClient that makes a request to create a topic on the broker?
Test Console App
using System;
using System.Diagnostics;
using System.Threading.Tasks;

using Confluent.Kafka;
using Confluent.Kafka.Admin;
using Polly;
using Polly.Retry;

namespace KafkaTransientFaultHandler
{
    class MainClass
    {
        private static AsyncRetryPolicy BrokerWaitAndRetry(short retryCount, short waitSeconds)
        {
            var pauseBetweenFailures = TimeSpan.FromSeconds(waitSeconds);

            var retryPolicy = Policy
                .Handle<KafkaException>()
                .WaitAndRetryAsync(retryCount, i => pauseBetweenFailures, onRetry: (Exception e, TimeSpan time) =>
                {
                    Console.WriteLine($"ONRETRY :: Waited for timepsan {time}");
                });

            return retryPolicy;
        }

        private static IAdminClient CreateAdminClient(string brokers, int socketTimeOutMs = 60000)
        {
            return new AdminClientBuilder(
                new AdminClientConfig()
                {
                    BootstrapServers = brokers,
                    Debug = "all",
                    SocketTimeoutMs = socketTimeOutMs,
                }
            )
                .Build();
        }

        private static async Task CreateTopic(IAdminClient client, string topic)
        {
            var stopWatch = new Stopwatch();

            stopWatch.Start();

            try
            {
                Console.WriteLine($"SENDING REQUEST TO CREATE TOPIC :: {topic}");

                await client.CreateTopicsAsync(
                   new TopicSpecification[] {
                        new TopicSpecification()
                        {
                            Name = topic,
                            NumPartitions = 3,
                            ReplicationFactor = 1
                        }
                   },
                   new CreateTopicsOptions
                   {
                       /** Set low request timeout for basic testing, otherwise would use a higher timeout of 60 seconds*/
                       RequestTimeout = TimeSpan.FromSeconds(20)
                   }
                );

                PrintRuntime(stopWatch.Elapsed);

                Console.WriteLine($"SUCCESSFULLY CREATED TOPIC :: {topic}");
            }
            catch (KafkaException ke)
            {
                Console.WriteLine($"EXCEPTION ENCOUNTERED CREATING TOPIC :: {topic}");
                Console.Error.WriteLine($"Kafka exception : {ke.Error.Code}");
                Console.Error.WriteLine($"Kafka error : {ke.Error.Reason}");
                Console.Error.WriteLine($"Kafka fatal : {ke.Error.IsFatal}");
                Console.Error.WriteLine($"Kafka broker : {ke.Error.IsBrokerError}");
                Console.Error.WriteLine($"Kafka local error : {ke.Error.IsLocalError}");

                PrintRuntime(stopWatch.Elapsed);

                throw ke;
            }
        }

        private static void PrintRuntime(TimeSpan ts)
        {
            string elapsedTime = String.Format("{0:00}:{1:00}:{2:00}.{3:00}",
                    ts.Hours, ts.Minutes, ts.Seconds,
                    ts.Milliseconds / 10);

            Console.WriteLine("RunTime " + elapsedTime);
        }

        private static void PrintUsage() => Console.WriteLine("Usage: .. <reuse|new> <broker,broker,..> <topic> [topic...]");

        /**
         * Small test console app to investigate transient failure handling for creating a topic using AdminClient when the broker is not online, i.e. unavailable.
         *
         * Usage:
         * ------
         * dotnet run reuse localhost:9092 mytopic:    Try and create a topic using a wait and retry failure handling strategy. On each retry reuse same admin client instance
         * dotnet run new localhost:9092 mytopic:      Try and create a topic using a wait and retry failure handling strategy. On each try use a new admin client instance
         *
         * Analysis:
         * ---------
         * When the same client instance is used to create a topic upon a retry attempt:    The program blocks when the rdkafka poll thread detects that a broker is connected and
         *                                                                                enters the main thread. The CreateTopicsAsync is not resumed, blocking occurs.
         *
         * When a new client instance is used to create a topic upon each retry attempt:    Success, the topic is created and the program exits.
         *
         *
         * Is the blocking behaviour happening because the rdkafka library hides/suppresses multiple instances of a local time out error?
         *
         *
         * How to test
         * -----------
         * Start this console app and wait until see console output informing that ONRETRY delegate has been triggered.
         * Then, start the kafka broker, e.g. from docker-compose stack
         *
         * The program will block when reusing the same client instance for each retry attempt.
         * The program will succeed when using a new client instance for each retry attempt.
         *
         * Questions
         * --------
         *
         * Does confluent Kafka SDK have built in support for transient failure handling for AdminClient when a broker is unavailable?
         * Is it possible to use the same client instance for retry attempts of CreateTopic operation, rather than create a new client instance each time?
         */
        public static async Task Main(string[] args)
        {
            if (args.Length < 3)
            {
                PrintUsage();
                return;
            }

            const short retryAttempts = 3;
            const short delayInterval = 2;

            var strategy = args[0];
            var brokers = args[1];
            var topic = args[2];

            Console.WriteLine($"Creating polly wait and retry transient failure policy with {retryAttempts} retries and a {delayInterval} second delay");
            var transientFailurePolicy = BrokerWaitAndRetry(retryAttempts, delayInterval);

            Console.WriteLine($"Creating admin client using broker(s) {brokers}, with a socket timeout of 20s for testing...");
            var client = CreateAdminClient(brokers, 20000);

            if(strategy.Equals("reuse")) {
                Console.WriteLine($"Trying to create topic {topic} using the wait and retry policy that reuses a client instance");
                await transientFailurePolicy.ExecuteAsync(async () => await CreateTopic(client, topic));
            }
            else if(strategy.Equals("new")) {
                Console.WriteLine($"Trying to create topic {topic} using the wait and retry policy that creates a new admin client for each retry attempt");
                await transientFailurePolicy.ExecuteAsync(async () =>
                {
                    client.Dispose();
                    client = CreateAdminClient(brokers, 20000);
                    await CreateTopic(client, topic);

                    Console.WriteLine($"TRANSIENT FAILURE STRATEGY SUCCESSFUL!");
                });
            }
            else
            {
                Console.Error.WriteLine("Unrecognised admin client strategy");
            }
        }
    }
}

Is it possible to configure rdkafka so that identical errors are not suppressed when trying to connect the AdminClient to the broker?

those are just debug logs. you can get an "all brokers down" error event (really an "info" level event - misnamed) by setting an error handler, but i'd just react to the exception thrown by CreateTopicAsync to determine success or not.  i wouldn't concern yourself with whether the client could connect to the broker or not. in general, there will be many brokers, and the logic about what happens when one goes down is complex (and encapsulated by the client). note that you'll have to consider the case where the broker request was sent before the a timeout, but the response wan't received. i.e. your application could get a timeout error in some cases where the topic was created. the next time you call you'll get a topic exists error (and you won't want to retry on that).

In the above example, once the rdkafka poll thread has eventually detected that the broker is up and running, why does the AdminClient create topics operation not complete?

the create topics function call should always complete (with success or exception) before or at the specified timeout. does it not?

How to use Confluent Kafka .NET with/without Polly to implement a retry and wait failure policy for a Kafka AdminClient that makes a request to create a topic on the broker?

you should be able to just call create topics with some reasonable timeout (10s?) and retry on exception that isn't topic already exists (i may have missed cases, haven't thought about it in depth).
i'm not familiar with polly. if your are still having problems, recommend taking polly out of the equation and demonstrating the issue using as simple as possible an application (i.e. just call CreateTopicsAsync, and if error, call it again). if that is causing an issue, i can take a look.
would recommend not fiddling with these parameters:
SocketTimeoutMs = 20000, // reduced from 60seconds for testing purposes
SocketMaxFails = 5, // is this synonymous to 5 retries?

SocketMaxFails is the number of timed-out broker requests (as determined by SocketTimeoutMs) that may occur before librdkafka closes the connection and re-opens it. it's highly recommended to leave that at 1.

Is it possible to configure rdkafka so that identical errors are not suppressed when trying to connect the AdminClient to the broker?

those are just debug logs. you can get an "all brokers down" error event (really an "info" level event - misnamed) by setting an error handler, but i'd just react to the exception thrown by CreateTopicAsync to determine success or not.  i wouldn't concern yourself with whether the client could connect to the broker or not. in general, there will be many brokers, and the logic about what happens when one goes down is complex (and encapsulated by the client). note that you'll have to consider the case where the broker request was sent before the a timeout, but the response wan't received. i.e. your application could get a timeout error in some cases where the topic was created. the next time you call you'll get a topic exists error (and you won't want to retry on that).

In the above example, once the rdkafka poll thread has eventually detected that the broker is up and running, why does the AdminClient create topics operation not complete?

the create topics function call should always complete (with success or exception) before or at the specified timeout. does it not?

How to use Confluent Kafka .NET with/without Polly to implement a retry and wait failure policy for a Kafka AdminClient that makes a request to create a topic on the broker?

you should be able to just call create topics with some reasonable timeout (10s?) and retry on exception that isn't topic already exists (i may have missed cases, haven't thought about it in depth).
i'm not familiar with polly. if your are still having problems, recommend taking polly out of the equation and demonstrating the issue using as simple as possible an application (i.e. just call CreateTopicsAsync, and if error, call it again). if that is causing an issue, i can take a look.
would recommend not fiddling with these parameters:
SocketTimeoutMs = 20000, // reduced from 60seconds for testing purposes
SocketMaxFails = 5, // is this synonymous to 5 retries?

SocketMaxFails is the number of timed-out broker requests (as determined by SocketTimeoutMs) that may occur before librdkafka closes the connection and re-opens it. it's highly recommended to leave that at 1.

Hi @mhowlett
Thanks for your patience and responding. Much appreciated!!

you should be able to just call create topics with some reasonable timeout (10s?) and retry on exception that isn't topic already exists (i may have missed cases, haven't thought about it in depth).

Ok, yep I will go with your suggestion of retry after request timeout and add isolating the topic already exists scenario. Many thanks for this suggestion.

the create topics function call should always complete (with success or exception) before or at the specified timeout. does it not?

On the first attempt, yes, after the specified timeout an exception is thrown. I am not familiar with what is happening under-the-hood. It looks as though there is an rdkafka poll thread still running when the AdminClient instance has thrown an exception. This is stopped when the client is disposed / program terminates.
The create topics (async) function blocks when issued a second time on the same AdminClient instance that previously threw the exception. If I start up the broker within the timeout period of the retry attempt, the rdkafka poll thread shows that it resumes to normal state after detecting that a broker is up and running. However, the create topics (async) function remains blocked.
Alternatively.....
Upon retrying, if I create a new AdminClient instance and issue a second create topics (async) request it is successful.
I have included a small console application, listed below that demonstrates the above scenarios.
In conclusion, I am guessing that for each retry attempt the recommended usage / semantics of the library is to create a new AdminClient instance and retry the create topics function. This is safer than operating on an instance that is in an exceptional state. Subsequently, it is the responsibility of the developer to then dispose of the exceptional instance, thus stopping the rdkafka poll thread and cleaning up other resources.
If this is the case, could some form of safety mechanism be implemented within the AdminClient? For example, a safety exception or the original first exception is thrown from the library when trying to invoke an operation on an AdminClient that is in an exceptional state.
Small console app for test
using System;
using System.Diagnostics;
using System.Threading.Tasks;

using Confluent.Kafka;
using Confluent.Kafka.Admin;

namespace KafkaTransientFaultHandler
{
    class MainClass
    {
        private static IAdminClient CreateAdminClient(string brokers, int socketTimeOutMs = 60000)
        {
            return new AdminClientBuilder(
                new AdminClientConfig()
                {
                    BootstrapServers = brokers,
                    Debug = "all",
                    SocketTimeoutMs = socketTimeOutMs,
                }
            )
                .Build();
        }

        private static async Task CreateTopic(IAdminClient client, string topic)
        {
            var stopWatch = new Stopwatch();

            stopWatch.Start();

            var topics = new TopicSpecification[]
            {
                new TopicSpecification()
                {
                    Name = topic,
                    NumPartitions = 3,
                    ReplicationFactor = 1
                }
            };

            var opts = new CreateTopicsOptions
            {
                RequestTimeout = TimeSpan.FromSeconds(10)
            };

            try
            {
                Console.WriteLine($"SENDING REQUEST TO CREATE TOPIC :: {topic}");

                await client.CreateTopicsAsync(topics, opts);

                PrintRuntime(stopWatch.Elapsed);

                Console.WriteLine($"SUCCESSFULLY CREATED TOPIC :: {topic}");
            }
            catch (KafkaException ke)
            {
                Console.WriteLine($"EXCEPTION ENCOUNTERED CREATING TOPIC 1ST REQUEST :: {topic}");
                Console.Error.WriteLine($"Kafka exception : {ke.Error.Code}");
                Console.Error.WriteLine($"Kafka error : {ke.Error.Reason}");
                Console.Error.WriteLine($"Kafka fatal : {ke.Error.IsFatal}");
                Console.Error.WriteLine($"Kafka broker : {ke.Error.IsBrokerError}");
                Console.Error.WriteLine($"Kafka local error : {ke.Error.IsLocalError}");

                PrintRuntime(stopWatch.Elapsed);

                throw ke;
            }
        }

        private static void PrintRuntime(TimeSpan ts)
        {
            string elapsedTime = String.Format("{0:00}:{1:00}:{2:00}.{3:00}",
                    ts.Hours, ts.Minutes, ts.Seconds,
                    ts.Milliseconds / 10);

            Console.WriteLine("RunTime " + elapsedTime);
        }

        private static void PrintUsage() => Console.WriteLine("Usage: .. <reuse|new> <broker,broker,..> <topic> [topic...]");

        /**
         * Small test console app to investigate transient failure handling for creating a topic using AdminClient when the broker is not online, i.e. unavailable.
         *
         * Usage:
         * ------
         * dotnet run reuse localhost:9092 mytopic:    Try and create a topic using a wait and retry failure handling strategy. On retry reuse same admin client instance
         * dotnet run new localhost:9092 mytopic:      Try and create a topic using a wait and retry failure handling strategy. On retry use a new admin client instance
         *
         * Analysis:
         * ---------
         * When the same AdminClient instance is used to create a topic upon a retry attempt:   The rdkafka poll thread detects that a broker is connected and
         *                                                                                enters the main thread. The CreateTopicsAsync operation blocks when retried on the SAME ClientAdmin instance.
         *
         *
         * When a new AdminClient instance is used to create a topic upon a retry attempt:      Success, the topic is created after a retry operation on a NEW ClientAdmin instance and the program exits.
         *
         *
         * Not sure what is happening under-the-hood with rdkafka poll thread.
         * Once the CreateTopicAsync operation has failed on the AdminClient it is an exceptional state and seems to be blocking. When the CreateTopic operation is retried on the SAME AdminClient instance the
         * debug logs show that the poll thread is still running and has detected that the broker is up and running. However, the CreateTopicAsyc operation blocks when retried on this SAME instance.
         * Is the blocking behaviour happening because the rdkafka library hides/suppresses multiple instances of a local time out error?
         *
         * In such a scenario for transient failure handling, the only way can get the retry to work is to dispose the old one(which has the rdkafka poll thread running back in normal non-exceptional state again)
         * and create a NEW client instance for the retry attempt.
         *
         * How to test
         * -----------
         * Start this console app and wait until see console output informing that retry has been triggered.
         * Then, start the kafka broker, e.g. from docker-compose stack
         *
         * The program will block when reusing the same client instance for each retry attempt (dotnet run reuse localhost:9092 mytopic)
         * The program will succeed when using a new client instance for each retry attempt (dotnet run new localhost:9092 mytopic)
         *
         * Questions
         * --------
         *
         * Is it possible to use the same AdminClient instance for a retry attempt(s) of CreateTopicAsync operation, rather than create a new AdminClient instance each time?
         */
        public static async Task Main(string[] args)
        {
            if (args.Length < 3)
            {
                PrintUsage();
                return;
            }

            var strategy = args[0];
            var brokers = args[1];
            var topic = args[2];

            Console.WriteLine($"Creating admin client using broker(s) {brokers}, with a socket timeout of 20s for testing...");
            var client = CreateAdminClient(brokers, 20000);

            try
            {
                // reuse the same client on retry, blocks => presumably because it is the instance is in am exceptional state. However, there is a poll background thread still running and the appliction blocks
                // Assuming that once client is in an exceptional state, it will block when it is reused? rdkafkalib is not throwing instances of the same error so Confluent Kafka .NET is not aware the exception occurred?
                if (strategy.Equals("reuse"))
                {
                    try
                    {
                        Console.WriteLine($"Trying to create topic {topic} using a simple wait and retry policy. On error it will retry the CreateTopic operation again using the same client instance");
                        await CreateTopic(client, topic);

                        Console.WriteLine($"CREATED TOPIC {topic} SUCCESSFULLY");
                    }
                    catch (KafkaException ke)
                    {
                        if(ke.Error.Code.Equals(ErrorCode.Local_TimedOut))
                        {
                            Console.WriteLine("WAITING 5 seconds BEFORE RETRY");
                            await Task.Delay(5000);

                            Console.WriteLine("RETRYING USING REUSE STRATEGY...");
                            Console.WriteLine($"Retrying creating topic {topic}. On error an exception will be rethrown and the application will exit");
                            await CreateTopic(client, topic);

                            Console.WriteLine($"CREATED TOPIC {topic} SUCCESSFULLY ON RETRY");
                        }
                    }
                }

                // create a new client instance on retry
                else if(strategy.Equals("new"))
                {
                    try
                    {
                        Console.WriteLine($"Trying to create topic {topic} using a simple wait and retry policy. On error it will retry the CreateTopic operation again using the same client instance");
                        await CreateTopic(client, topic);

                        Console.WriteLine($"CREATED TOPIC {topic} SUCCESSFULLY");
                    }
                    catch (KafkaException ke)
                    {
                        if (ke.Error.Code.Equals(ErrorCode.Local_TimedOut))
                        {
                            Console.WriteLine("Detected a request timeout failure, waiting 5 seconds before retrying");
                            await Task.Delay(5000);

                            Console.WriteLine("RETRYING USING NEW CLIENTADMIN INSTANCE...");
                            Console.WriteLine($"Creating a new AdminClient upon retry, using broker(s) {brokers}, with a socket timeout of 20s for testing...");
                            client.Dispose();
                            client = CreateAdminClient(brokers, 20000);

                            Console.WriteLine($"Retrying creating topic {topic}. On error an exception will be rethrown and the application will exit");
                            await CreateTopic(client, topic);

                            Console.WriteLine($"CREATED TOPIC {topic} SUCCESSFULLY ON RETRY");
                        }
                    }

                }
            }
            catch(Exception ex)
            {
                Console.Error.WriteLine($"An exception has been thrown :: {ex} after retry operations");
            }
            finally
            {
                // fail silently if already disposed
                try { client.Dispose(); } catch(ObjectDisposedException) { }
            }
        }
    }
}

Hi @mhowlett
Thanks for your patience and responding. Much appreciated!!

you should be able to just call create topics with some reasonable timeout (10s?) and retry on exception that isn't topic already exists (i may have missed cases, haven't thought about it in depth).

Ok, yep I will go with your suggestion of retry after request timeout and add isolating the topic already exists scenario. Many thanks for this suggestion.

the create topics function call should always complete (with success or exception) before or at the specified timeout. does it not?

On the first attempt, yes, after the specified timeout an exception is thrown. I am not familiar with what is happening under-the-hood. It looks as though there is an rdkafka poll thread still running when the AdminClient instance has thrown an exception. This is stopped when the client is disposed / program terminates.
The create topics (async) function blocks when issued a second time on the same AdminClient instance that previously threw the exception. If I start up the broker within the timeout period of the retry attempt, the rdkafka poll thread shows that it resumes to normal state after detecting that a broker is up and running. However, the create topics (async) function remains blocked.
Alternatively.....
Upon retrying, if I create a new AdminClient instance and issue a second create topics (async) request it is successful.
I have included a small console application, listed below that demonstrates the above scenarios.
In conclusion, I am guessing that for each retry attempt the recommended usage / semantics of the library is to create a new AdminClient instance and retry the create topics function. This is safer than operating on an instance that is in an exceptional state. Subsequently, it is the responsibility of the developer to then dispose of the exceptional instance, thus stopping the rdkafka poll thread and cleaning up other resources.
If this is the case, could some form of safety mechanism be implemented within the AdminClient? For example, a safety exception or the original first exception is thrown from the library when trying to invoke an operation on an AdminClient that is in an exceptional state.
Small console app for test
using System;
using System.Diagnostics;
using System.Threading.Tasks;

using Confluent.Kafka;
using Confluent.Kafka.Admin;

namespace KafkaTransientFaultHandler
{
    class MainClass
    {
        private static IAdminClient CreateAdminClient(string brokers, int socketTimeOutMs = 60000)
        {
            return new AdminClientBuilder(
                new AdminClientConfig()
                {
                    BootstrapServers = brokers,
                    Debug = "all",
                    SocketTimeoutMs = socketTimeOutMs,
                }
            )
                .Build();
        }

        private static async Task CreateTopic(IAdminClient client, string topic)
        {
            var stopWatch = new Stopwatch();

            stopWatch.Start();

            var topics = new TopicSpecification[]
            {
                new TopicSpecification()
                {
                    Name = topic,
                    NumPartitions = 3,
                    ReplicationFactor = 1
                }
            };

            var opts = new CreateTopicsOptions
            {
                RequestTimeout = TimeSpan.FromSeconds(10)
            };

            try
            {
                Console.WriteLine($"SENDING REQUEST TO CREATE TOPIC :: {topic}");

                await client.CreateTopicsAsync(topics, opts);

                PrintRuntime(stopWatch.Elapsed);

                Console.WriteLine($"SUCCESSFULLY CREATED TOPIC :: {topic}");
            }
            catch (KafkaException ke)
            {
                Console.WriteLine($"EXCEPTION ENCOUNTERED CREATING TOPIC 1ST REQUEST :: {topic}");
                Console.Error.WriteLine($"Kafka exception : {ke.Error.Code}");
                Console.Error.WriteLine($"Kafka error : {ke.Error.Reason}");
                Console.Error.WriteLine($"Kafka fatal : {ke.Error.IsFatal}");
                Console.Error.WriteLine($"Kafka broker : {ke.Error.IsBrokerError}");
                Console.Error.WriteLine($"Kafka local error : {ke.Error.IsLocalError}");

                PrintRuntime(stopWatch.Elapsed);

                throw ke;
            }
        }

        private static void PrintRuntime(TimeSpan ts)
        {
            string elapsedTime = String.Format("{0:00}:{1:00}:{2:00}.{3:00}",
                    ts.Hours, ts.Minutes, ts.Seconds,
                    ts.Milliseconds / 10);

            Console.WriteLine("RunTime " + elapsedTime);
        }

        private static void PrintUsage() => Console.WriteLine("Usage: .. <reuse|new> <broker,broker,..> <topic> [topic...]");

        /**
         * Small test console app to investigate transient failure handling for creating a topic using AdminClient when the broker is not online, i.e. unavailable.
         *
         * Usage:
         * ------
         * dotnet run reuse localhost:9092 mytopic:    Try and create a topic using a wait and retry failure handling strategy. On retry reuse same admin client instance
         * dotnet run new localhost:9092 mytopic:      Try and create a topic using a wait and retry failure handling strategy. On retry use a new admin client instance
         *
         * Analysis:
         * ---------
         * When the same AdminClient instance is used to create a topic upon a retry attempt:   The rdkafka poll thread detects that a broker is connected and
         *                                                                                enters the main thread. The CreateTopicsAsync operation blocks when retried on the SAME ClientAdmin instance.
         *
         *
         * When a new AdminClient instance is used to create a topic upon a retry attempt:      Success, the topic is created after a retry operation on a NEW ClientAdmin instance and the program exits.
         *
         *
         * Not sure what is happening under-the-hood with rdkafka poll thread.
         * Once the CreateTopicAsync operation has failed on the AdminClient it is an exceptional state and seems to be blocking. When the CreateTopic operation is retried on the SAME AdminClient instance the
         * debug logs show that the poll thread is still running and has detected that the broker is up and running. However, the CreateTopicAsyc operation blocks when retried on this SAME instance.
         * Is the blocking behaviour happening because the rdkafka library hides/suppresses multiple instances of a local time out error?
         *
         * In such a scenario for transient failure handling, the only way can get the retry to work is to dispose the old one(which has the rdkafka poll thread running back in normal non-exceptional state again)
         * and create a NEW client instance for the retry attempt.
         *
         * How to test
         * -----------
         * Start this console app and wait until see console output informing that retry has been triggered.
         * Then, start the kafka broker, e.g. from docker-compose stack
         *
         * The program will block when reusing the same client instance for each retry attempt (dotnet run reuse localhost:9092 mytopic)
         * The program will succeed when using a new client instance for each retry attempt (dotnet run new localhost:9092 mytopic)
         *
         * Questions
         * --------
         *
         * Is it possible to use the same AdminClient instance for a retry attempt(s) of CreateTopicAsync operation, rather than create a new AdminClient instance each time?
         */
        public static async Task Main(string[] args)
        {
            if (args.Length < 3)
            {
                PrintUsage();
                return;
            }

            var strategy = args[0];
            var brokers = args[1];
            var topic = args[2];

            Console.WriteLine($"Creating admin client using broker(s) {brokers}, with a socket timeout of 20s for testing...");
            var client = CreateAdminClient(brokers, 20000);

            try
            {
                // reuse the same client on retry, blocks => presumably because it is the instance is in am exceptional state. However, there is a poll background thread still running and the appliction blocks
                // Assuming that once client is in an exceptional state, it will block when it is reused? rdkafkalib is not throwing instances of the same error so Confluent Kafka .NET is not aware the exception occurred?
                if (strategy.Equals("reuse"))
                {
                    try
                    {
                        Console.WriteLine($"Trying to create topic {topic} using a simple wait and retry policy. On error it will retry the CreateTopic operation again using the same client instance");
                        await CreateTopic(client, topic);

                        Console.WriteLine($"CREATED TOPIC {topic} SUCCESSFULLY");
                    }
                    catch (KafkaException ke)
                    {
                        if(ke.Error.Code.Equals(ErrorCode.Local_TimedOut))
                        {
                            Console.WriteLine("WAITING 5 seconds BEFORE RETRY");
                            await Task.Delay(5000);

                            Console.WriteLine("RETRYING USING REUSE STRATEGY...");
                            Console.WriteLine($"Retrying creating topic {topic}. On error an exception will be rethrown and the application will exit");
                            await CreateTopic(client, topic);

                            Console.WriteLine($"CREATED TOPIC {topic} SUCCESSFULLY ON RETRY");
                        }
                    }
                }

                // create a new client instance on retry
                else if(strategy.Equals("new"))
                {
                    try
                    {
                        Console.WriteLine($"Trying to create topic {topic} using a simple wait and retry policy. On error it will retry the CreateTopic operation again using the same client instance");
                        await CreateTopic(client, topic);

                        Console.WriteLine($"CREATED TOPIC {topic} SUCCESSFULLY");
                    }
                    catch (KafkaException ke)
                    {
                        if (ke.Error.Code.Equals(ErrorCode.Local_TimedOut))
                        {
                            Console.WriteLine("Detected a request timeout failure, waiting 5 seconds before retrying");
                            await Task.Delay(5000);

                            Console.WriteLine("RETRYING USING NEW CLIENTADMIN INSTANCE...");
                            Console.WriteLine($"Creating a new AdminClient upon retry, using broker(s) {brokers}, with a socket timeout of 20s for testing...");
                            client.Dispose();
                            client = CreateAdminClient(brokers, 20000);

                            Console.WriteLine($"Retrying creating topic {topic}. On error an exception will be rethrown and the application will exit");
                            await CreateTopic(client, topic);

                            Console.WriteLine($"CREATED TOPIC {topic} SUCCESSFULLY ON RETRY");
                        }
                    }

                }
            }
            catch(Exception ex)
            {
                Console.Error.WriteLine($"An exception has been thrown :: {ex} after retry operations");
            }
            finally
            {
                // fail silently if already disposed
                try { client.Dispose(); } catch(ObjectDisposedException) { }
            }
        }
    }
}

I am guessing that for each retry attempt the recommended usage / semantics of the library is to create a new AdminClient instance and retry the create topics function.

no, it's recommended you avoid creating clients wherever possible (it is not cheap).
so to clarify, you are saying that in the "reuse" case, you are seeing:

WAITING 5 seconds BEFORE RETRY
RETRYING USING REUSE STRATEGY...
Retrying creating topic {topic}. On error an exception will be rethrown and the application will exit

... and then the program blocks? (no exception message or topic created message)
if that is the case, i'll try and replicate, because that's not expected.

I am guessing that for each retry attempt the recommended usage / semantics of the library is to create a new AdminClient instance and retry the create topics function.

no, it's recommended you avoid creating clients wherever possible (it is not cheap).
so to clarify, you are saying that in the "reuse" case, you are seeing:

WAITING 5 seconds BEFORE RETRY
RETRYING USING REUSE STRATEGY...
Retrying creating topic {topic}. On error an exception will be rethrown and the application will exit

... and then the program blocks? (no exception message or topic created message)
if that is the case, i'll try and replicate, because that's not expected.

Thanks @mhowlett
In "reuse" case with the code that I listed earlier I am seeing:

WAITING 5 seconds BEFORE RETRY
RETRYING USING REUSE STRATEGY...
Retrying creating topic {topic}. On error an exception will be rethrown and the application will exit

...At this point I then start up the broker, e.g. via a docker-compose environment, well within the timeout period.
Subsequently, rdkafka library eventually sees that the broker is up and running and enters normal state in it's main thread. Then the program blocks.
It could be something that I doing incorrectly in the code or something incorrectly configured in Confluent Kafka 6.0.1
docker-compose environment. I am using Confluent Kafka .NET 1.5.3.
Do you want me to send the docker-compose file contents, just in case?

Thanks @mhowlett
In "reuse" case with the code that I listed earlier I am seeing:

WAITING 5 seconds BEFORE RETRY
RETRYING USING REUSE STRATEGY...
Retrying creating topic {topic}. On error an exception will be rethrown and the application will exit

...At this point I then start up the broker, e.g. via a docker-compose environment, well within the timeout period.
Subsequently, rdkafka library eventually sees that the broker is up and running and enters normal state in it's main thread. Then the program blocks.
It could be something that I doing incorrectly in the code or something incorrectly configured in Confluent Kafka 6.0.1
docker-compose environment. I am using Confluent Kafka .NET 1.5.3.
Do you want me to send the docker-compose file contents, just in case?

oh, and you should also see SENDING REQUEST TO CREATE TOPIC :: {topic}, right? then it blocks? and you don't see the output of PrintRuntime or "SUCCESSFULLY CREATED TOPIC :: {topic}.
await client.CreateTopicsAsync(topics, opts); should never block for longer than the timeout specified. i would be surprised to learn that is happening, it's a bug if so.
finally, can you make sure you leave the method blocking for a long time (say >10 minutes). after that i'll try to replicate.
you may be mucking up some details with docker (people commonly do, this might be helpful: https://www.confluent.io/blog/kafka-listeners-explained), however there is nothing that would make the CreateTopicsAsync method block indefinitely.

oh, and you should also see SENDING REQUEST TO CREATE TOPIC :: {topic}, right? then it blocks? and you don't see the output of PrintRuntime or "SUCCESSFULLY CREATED TOPIC :: {topic}.
await client.CreateTopicsAsync(topics, opts); should never block for longer than the timeout specified. i would be surprised to learn that is happening, it's a bug if so.
finally, can you make sure you leave the method blocking for a long time (say >10 minutes). after that i'll try to replicate.
you may be mucking up some details with docker (people commonly do, this might be helpful: https://www.confluent.io/blog/kafka-listeners-explained), however there is nothing that would make the CreateTopicsAsync method block indefinitely.

Many thanks @mhowlett

oh, and you should also see SENDING REQUEST TO CREATE TOPIC :: {topic}, right? then it blocks? and you don't see the output of PrintRuntime or "SUCCESSFULLY CREATED TOPIC :: {topic}.

Yes, correct, seeing SENDING REQUEST TO CREATE TOPIC :: {topic} and then it blocks. No output of CREATED TOPIC {topic} SUCCESSFULLY ON RETRY. However the debug logs (included below) show that rdkafka has:
sent CreateTopicRequest
received CreateTopicResponse

The debug logs were sampled over a 20 minute time period. SUCCESSFULLY CREATED TOPIC :: {topic} is not output and the program continues to block.

you may be mucking up some details with docker (people commonly do, this might be helpful: https://www.confluent.io/blog/kafka-listeners-explained), however there is nothing that would make the CreateTopicsAsync method block indefinitely.

Under normal conditions the program works as expected with the SUCCESSFULLY CREATED TOPIC :: {topic} message displayed. Just in case it is docker related, I have also included the docker-compose file contents and the docker-compose logs below.
Debug Logs when running reuse strategy
%7|1609677551.873|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature BrokerBalancedConsumer: LeaveGroup (0..0) supported by broker
%7|1609677551.873|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1: Enabling feature BrokerBalancedConsumer
%7|1609677551.873|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature ThrottleTime: Produce (1..2) supported by broker
%7|1609677551.873|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature ThrottleTime: Fetch (1..2) supported by broker
%7|1609677551.873|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1: Enabling feature ThrottleTime
%7|1609677551.873|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature Sasl: JoinGroup (0..0) supported by broker
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1: Enabling feature Sasl
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature SaslHandshake: SaslHandshake (0..0) supported by broker
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1: Enabling feature SaslHandshake
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature LZ4: FindCoordinator (0..0) supported by broker
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1: Enabling feature LZ4
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature OffsetTime: Offset (1..1) supported by broker
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1: Enabling feature OffsetTime
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature IdempotentProducer: InitProducerId (0..0) supported by broker
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1: Enabling feature IdempotentProducer
%7|1609677551.874|APIVERSION|rdkafka#producer-1| [thrd:<IP_ADDRESS>:9092/1]: <IP_ADDRESS>:9092/1:  Feature ZSTD: Produce (7..7) supported by broker
Creating admin client using broker(s) localhost:9092, with a socket timeout of 20s for testing...
%7|1609678365.835|WAKEUPFD|rdkafka#producer-1| [thrd:app]: localhost:9092/bootstrap: Enabled low-latency ops queue wake-ups
%7|1609678365.835|BRKMAIN|rdkafka#producer-1| [thrd::0/internal]: :0/internal: Enter main broker thread
%7|1609678365.835|BROKER|rdkafka#producer-1| [thrd:app]: localhost:9092/bootstrap: Added new broker with NodeId -1
%7|1609678365.835|CONNECT|rdkafka#producer-1| [thrd:app]: localhost:9092/bootstrap: Selected for cluster connection: bootstrap servers added (broker
has 0 connection attempt(s))
%7|1609678365.835|BRKMAIN|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Enter main broker thread
%7|1609678365.835|INIT|rdkafka#producer-1| [thrd:app]: librdkafka v1.5.3 (0x10503ff) rdkafka#producer-1 initialized (builtin.features gzip,snappy,ssl
,sasl,regex,lz4,sasl_gssapi,sasl_plain,sasl_scram,plugins,zstd,sasl_oauthbearer, STATIC_LINKING CC GXX PKGCONFIG OSXLD LIBDL PLUGINS ZLIB SSL SASL_CY
RUS ZSTD HDRHISTOGRAM SNAPPY SOCKEM SASL_SCRAM SASL_OAUTHBEARER CRC32C_HW, debug 0xfffff)
%7|1609678365.835|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678365.835|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678365.835|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678365.836|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678365.836|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678365.836|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678365.838|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) wi
th socket 75
%7|1609678365.838|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection r
efused (after 2ms in state CONNECT) (_TRANSPORT)
%3|1609678365.838|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection r
efused (after 2ms in state CONNECT)
%7|1609678365.838|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%3|1609678365.838|ERROR|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: 1/1 brokers are down
%7|1609678365.838|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678365.838|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678365.838|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678365.838|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678365.839|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678365.839|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%3|1609678365.842|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection r
efused (after 2ms in state CONNECT)
Trying to create topic foo using a simple wait and retry policy. On error it will retry the CreateTopic operation again using the same client instanc
e
SENDING REQUEST TO CREATE TOPIC :: foo
%7|1609678365.852|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state initializing: Success
%7|1609678365.852|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678365.852|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 33ms: lookup control
ler
%7|1609678366.835|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker h
as 1 connection attempt(s))
%7|1609678366.835|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678366.835|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678366.835|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678366.835|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678366.835|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678366.835|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678366.835|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678366.835|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678366.835|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678366.836|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 75
%7|1609678366.836|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678366.836|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678366.836|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678366.836|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678366.836|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%3|1609678366.836|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678366.836|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678366.836|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678366.836|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678366.836|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678366.836|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678366.836|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678366.836|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678366.836|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678367.837|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 2 connection attempt(s))
%7|1609678367.837|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678367.837|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678367.837|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678367.837|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678367.837|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678367.837|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678367.837|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678367.837|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678367.837|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678367.838|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) with socket 75
%7|1609678367.838|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678367.838|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678367.838|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678367.838|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678367.838|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678367.838|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678367.838|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678367.838|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678367.838|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678367.838|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678367.838|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678367.838|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%3|1609678367.838|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)
%7|1609678368.839|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 3 connection attempt(s))
%7|1609678368.839|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678368.839|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678368.839|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678368.839|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678368.839|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678368.839|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678368.839|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678368.839|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678368.839|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678368.840|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) with socket 75
%7|1609678368.840|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed) (_TRANSPORT): identical to last error
%3|1609678368.840|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
%7|1609678368.840|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678368.840|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678368.840|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678368.840|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%3|1609678368.840|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, 1 identical error(s) suppressed)
%7|1609678368.840|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678368.841|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678368.841|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678368.841|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678368.841|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678368.841|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678368.841|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678369.840|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 4 connection attempt(s))
%7|1609678369.841|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678369.841|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678369.841|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678369.841|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678369.841|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678369.841|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678369.841|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678369.841|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678369.841|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678369.841|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) with socket 75
%7|1609678369.841|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT): identical to last error: error log suppressed
%7|1609678369.841|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678369.841|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678369.841|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678369.841|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678369.841|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678369.841|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678369.841|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678369.841|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678369.842|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678369.842|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678369.842|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678369.842|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678369.842|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678370.842|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 5 connection attempt(s))
%7|1609678370.842|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678370.842|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678370.842|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678370.842|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678370.842|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678370.842|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678370.842|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678370.842|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678370.842|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678370.843|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 75
%7|1609678370.843|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678370.843|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678370.843|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%3|1609678370.843|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678370.843|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678370.843|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678370.843|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678370.843|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678370.843|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678370.843|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678370.843|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678370.844|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678370.844|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678370.844|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678371.844|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 6 connection attempt(s))
%7|1609678371.844|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678371.844|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678371.844|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678371.844|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678371.844|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678371.844|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678371.844|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678371.844|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678371.844|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678371.845|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678371.845|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 75
%7|1609678371.845|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT): identical to last error: error log suppressed
%7|1609678371.845|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678371.845|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678371.845|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678371.845|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678371.845|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678371.845|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678371.845|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678371.845|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678371.845|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678371.845|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678371.845|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678372.848|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 7 connection attempt(s))
%7|1609678372.848|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678372.848|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678372.848|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678372.848|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678372.848|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678372.848|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678372.848|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678372.848|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678372.848|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678372.849|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) with socket 75
%7|1609678372.849|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678372.849|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678372.849|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678372.849|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678372.849|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678372.849|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678372.850|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678372.849|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678372.850|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678372.850|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678372.850|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%3|1609678372.849|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)
%7|1609678372.850|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678372.850|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678373.850|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 8 connection attempt(s))
%7|1609678373.850|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678373.850|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678373.850|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678373.850|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678373.850|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678373.850|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678373.850|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678373.850|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678373.850|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678373.851|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 75
%7|1609678373.851|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678373.851|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678373.851|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678373.851|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678373.851|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678373.851|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678373.851|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678373.851|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%3|1609678373.851|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678373.851|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678373.851|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678373.851|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678373.851|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678373.852|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678374.850|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 9 connection attempt(s))
%7|1609678374.850|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678374.850|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678374.850|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678374.850|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678374.850|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678374.850|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678374.850|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678374.851|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678374.851|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678374.851|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) with socket 75
%7|1609678374.851|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678374.851|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678374.851|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678374.851|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%3|1609678374.851|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)
%7|1609678374.851|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678374.851|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678374.851|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678374.851|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678374.852|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678374.852|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678374.852|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678374.852|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678374.852|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678374.852|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678374.852|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678375.850|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 10 connection attempt(s))
%7|1609678375.850|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678375.850|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678375.850|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678375.850|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678375.850|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678375.851|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678375.851|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678375.851|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678375.851|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678375.851|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 75
%7|1609678375.851|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678375.851|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678375.851|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678375.851|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678375.851|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678375.851|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678375.851|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678375.852|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678375.851|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%3|1609678375.851|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678375.852|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678375.852|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678375.852|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678375.852|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678375.852|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Local: Timed out
%7|1609678375.852|ADMINFAIL|rdkafka#producer-1| [thrd:main]: Admin CREATETOPICS result error: Failed while waiting for controller: Local: Timed out
EXCEPTION ENCOUNTERED CREATING TOPIC 1ST REQUEST :: foo
Kafka exception : Local_TimedOut
Kafka error : Failed while waiting for controller: Local: Timed out
Kafka fatal : False
Kafka broker : False
Kafka local error : True
RunTime 00:00:10.15
WAITING 5 seconds BEFORE RETRY
%7|1609678376.851|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 11 connection attempt(s))
%7|1609678376.851|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678376.851|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678376.851|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678376.851|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678376.851|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678376.851|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678376.852|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 76
%7|1609678376.852|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT): identical to last error: error log suppressed
%7|1609678376.852|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678376.852|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678376.852|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678376.852|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678376.852|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678376.852|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678376.852|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678377.852|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 12 connection attempt(s))
%7|1609678377.852|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678377.852|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678377.852|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678377.852|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678377.852|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678377.852|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678377.852|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 76
%7|1609678377.853|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT): identical to last error: error log suppressed
%7|1609678377.853|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678377.853|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678377.853|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678377.853|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678377.853|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678377.853|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678377.853|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678378.853|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 13 connection attempt(s))
%7|1609678378.853|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678378.853|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678378.853|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678378.853|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678378.853|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678378.853|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678378.854|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 76
%7|1609678378.854|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT): identical to last error: error log suppressed
%7|1609678378.854|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678378.854|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678378.854|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678378.854|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678378.854|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678378.854|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678378.854|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678379.855|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 14 connection attempt(s))
%7|1609678379.855|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678379.855|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678379.855|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678379.855|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678379.855|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678379.855|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678379.855|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) with socket 76
%7|1609678379.855|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678379.855|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678379.855|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678379.855|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678379.855|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%3|1609678379.855|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)
%7|1609678379.856|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678379.856|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678379.856|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678379.856|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678380.856|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 15 connection attempt(s))
%7|1609678380.856|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678380.857|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678380.857|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678380.857|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678380.857|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678380.857|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678380.857|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv4#<IP_ADDRESS>:9092 (plaintext) with socket 76
%7|1609678380.857|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678380.857|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678380.857|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678380.857|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678380.857|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678380.857|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678380.857|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678380.857|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678380.857|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%3|1609678380.857|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#<IP_ADDRESS>:9092 failed: Connection refused (after 0ms in state CONNECT)
RETRYING USING REUSE STRATEGY...
Retrying creating topic foo. On error an exception will be rethrown and the application will exit
SENDING REQUEST TO CREATE TOPIC :: foo
%7|1609678381.149|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state initializing: Success
%7|1609678381.149|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678381.149|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: lookup controller (broker has 16 connection attempt(s))
%7|1609678381.149|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678381.149|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678381.149|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678381.150|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678381.150|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678381.150|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678381.150|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678381.150|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678381.150|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678381.150|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) with socket 76
%7|1609678381.150|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT)
%3|1609678381.150|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT)
%7|1609678381.151|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678381.151|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678381.151|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678381.151|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%3|1609678381.151|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)
%7|1609678381.151|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678381.151|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678381.151|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678381.151|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678381.151|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678381.151|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678381.151|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678381.151|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678381.151|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 48ms: lookup controller
%7|1609678381.858|CONNECT|rdkafka#producer-1| [thrd:main]: localhost:9092/bootstrap: Selected for cluster connection: no cluster connection (broker has 17 connection attempt(s))
%7|1609678381.858|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Received CONNECT op
%7|1609678381.858|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1609678381.858|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678381.858|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state TRY_CONNECT connecting
%7|1609678381.858|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1609678381.858|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678381.858|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678381.858|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678381.858|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678381.859|CONNECT|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to ipv6#[::1]:9092 (plaintext) with socket 76
%7|1609678381.859|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT) (_TRANSPORT): identical to last error: error log suppressed
%7|1609678381.859|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state CONNECT -> DOWN
%7|1609678381.859|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1609678381.859|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678381.859|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1609678381.859|CREATETOPICS|rdkafka#producer-1| [thrd:main]: CREATETOPICS worker called in state waiting for controller: Success
%7|1609678381.859|ADMIN|rdkafka#producer-1| [thrd:main]: CREATETOPICS: looking up controller
%7|1609678381.859|CONNECT|rdkafka#producer-1| [thrd:main]: Not selecting any broker for cluster connection: still suppressed for 49ms: lookup controller
%7|1609678381.859|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1609678381.859|STATE|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed state DOWN -> INIT
%7|1609678381.859|BROADCAST|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: Broadcasting state change