Сброс состояния Kafka

Редактировал(а) Alexandr Fokin 2024/07/01 11:01
 public static async Task CleanAsync(
 KafkaContainer container
 )
{    
   var connectionString = /* setup connection */;
   using var client = new AdminClientBuilder(
       new AdminClientConfig()
        {
            BootstrapServers = connectionString
        }
        )
        .Build();

   var metadata = client.GetMetadata(TimeSpan.FromSeconds(5));
   var topics = metadata.Topics
        .Select(e => e.Topic)
        .Where(e => !e.StartsWith("__")) // Не трогаем системные топики
        .ToArray();

   if (topics.Length != 0)
    {
       await client.DeleteTopicsAsync(
            topics
            );
    }    
}
 Testcontainers, Containers pool
 
Теги: