Как протестировать потребителя кафки против реального брокера кафки, работающего на сервере?

Question

У меня возникают трудности с пониманием некоторых концепций Kafka в Java Spring Boot. Я хотел бы протестировать потребителя с реальным брокером Kafka, работающим на сервере, у которого есть несколько производителей, которые пишут / уже писали данные в различные темы. Я хотел бы установить соединение с сервером, использовать данные и проверить или обработать их содержимое в тесте.

Подавляющее большинство примеров (на самом деле все, что я видел до сих пор) в Интернете относятся к встроенной kafka, EmbeddedKafkaBroker и показывают как производителя, так и потребителя, реализованного на одной машине локально. Я не нашел ни одного примера, который объяснил бы, как установить соединение с удаленным сервером kafka и прочитать данные из определенной темы. Я написал некоторый код и напечатал адрес брокера:

System.out.println(embeddedKafkaBroker.getBrokerAddress(0));

Я получил 127.0.0.1:9092, что означает, что он локальный, поэтому соединение с удаленным сервером не установлено.

С другой стороны, когда я запускаю SpringBootApplication, я получаю полезную нагрузку от удаленного брокера.

Получатель:

@Component
public class Receiver {

private static final String TOPIC_NAME = "X";

private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class);

private CountDownLatch latch = new CountDownLatch(1);

public CountDownLatch getLatch() {
    return latch;
}

@KafkaListener(topics = TOPIC_NAME)
public void receive(final byte[] payload) {
    LOGGER.info("received the following payload: '{}'", payload);
    latch.countDown();
}
}

Config:

    @EnableKafka
    @Configuration
    public class ByteReceiverConfig {

        @Autowired
        EmbeddedKafkaBroker kafkaEmbeded;

        @Value("${spring.kafka.bootstrap-servers}")
        private String bootstrapServers;

        @Value("${spring.kafka.consumer.group-id}")
        private String groupIdConfig;

        @Bean
        public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
            final ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
                    new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            return factory;
        }

        @Bean
        ConsumerFactory<Object, Object> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerProperties());
        }

        @Bean
        Map<String, Object> consumerProperties() {
            final Map<String, Object> properties =
                    KafkaTestUtils.consumerProps("junit-test", "true", this.kafkaEmbeded);
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);
            return properties;
        }

Тестовое задание:

        @EnableAutoConfiguration
        @EnableKafka
        @SpringBootTest(classes = {ByteReceiverConfig.class, Receiver.class})
        @EmbeddedKafka
        @ContextConfiguration(classes = ByteReceiverConfig.class)
        @TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
                "spring.kafka.consumer.group-id=EmbeddedKafkaTest"})
        public class KafkaTest {


            @Autowired
            private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

            @Autowired
            EmbeddedKafkaBroker embeddedKafkaBroker;


            @Autowired
            Receiver receiver;

            @BeforeEach
            void waitForAssignment() {
                for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
                    System.out.println(messageListenerContainer.getAssignedPartitions().isEmpty());
                    System.out.println(messageListenerContainer.toString());
                    System.out.println(embeddedKafkaBroker.getTopics().size());
                    System.out.println(embeddedKafkaBroker.getPartitionsPerTopic());
                    System.out.println(embeddedKafkaBroker.getBrokerAddress(0));
                    System.out.println(embeddedKafkaBroker.getBrokersAsString());

                    ContainerTestUtils.waitForAssignment(messageListenerContainer,
                            embeddedKafkaBroker.getPartitionsPerTopic());
            }

            @Test
            public void testReceive() {

            }
        }

Я хотел бы, чтобы кто-то пролил свет на следующие вопросы:

1. Может ли экземпляр класса EmbeddedKafkaBroker использоваться для тестирования данных, которые поступают от удаленного брокера, или он используется только для локальных тестов, в которых я бы прокументировал, то есть отправил данные в созданную мной тему и сам использовал данные?

2. Можно ли написать тестовый класс для реального сервера kafka? Например, чтобы проверить, было ли установлено соединение, или данные были прочитаны из определенной темы. Какие аннотации, конфигурации и классы понадобятся в таком случае?

3.Если я хочу использовать только данные, нужно ли указывать конфигурацию производителя в файле конфигурации (это было бы странно, но все примеры, с которыми я сталкивался до сих пор), сделали это?

4. Знаете ли вы какие-либо ресурсы (книги, веб-сайты и т. Д.), Которые показывают реальные примеры использования kafka, то есть с удаленным сервером kafka, только с прокудером или потребителем?

Answer 1

1. Вам вообще не нужен встроенный брокер, если вы хотите общаться только с внешним брокером.

2. Да, просто правильно установите свойство серверов начальной загрузки.

3. Нет, вам не нужна конфигурация производителя.

РЕДАКТИРОВАТЬ

@SpringBootApplication
public class So56044105Application {

    public static void main(String[] args) {
        SpringApplication.run(So56044105Application.class, args);
    }

    @Bean
    public NewTopic topic() {
        return new NewTopic("so56044105", 1, (short) 1);
    }

}

spring.kafka.bootstrap-servers=10.0.0.8:9092
spring.kafka.consumer.enable-auto-commit=false

@RunWith(SpringRunner.class)
@SpringBootTest(classes = { So56044105Application.class, So56044105ApplicationTests.Config.class })
public class So56044105ApplicationTests {

    @Autowired
    public Config config;

    @Test
    public void test() throws InterruptedException {
        assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue();
        assertThat(config.received.get(0)).isEqualTo("foo");
    }

    @Configuration
    public static class Config implements ConsumerSeekAware {

        List<String> received = new ArrayList<>();

        CountDownLatch latch = new CountDownLatch(3);

        @KafkaListener(id = "so56044105", topics = "so56044105")
        public void listen(String in) {
            System.out.println(in);
            this.received.add(in);
            this.latch.countDown();
        }

        @Override
        public void registerSeekCallback(ConsumerSeekCallback callback) {
        }

        @Override
        public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
            System.out.println("Seeking to beginning");
            assignments.keySet().forEach(tp -> callback.seekToBeginning(tp.topic(), tp.partition()));
        }

        @Override
        public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        }

    }

}

Answer 2

В этом репозитории есть несколько примеров начальной загрузки реальных производителей и потребителей Kafka в различных конфигурациях — открытый текст, SSL, с аутентификацией и без нее и т. д.

Примечание: приведенный выше репозиторий содержит примеры для книги « Эффективный Кафка », автором которой я являюсь. Однако их можно свободно использовать без книги, и, надеюсь, они имеют такой же смысл сами по себе.

Более того, вот пара примеров для основного производителя и потребителя.

      /** A sample Kafka producer. */
import static java.lang.System.*;

import java.util.*;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.*;

public final class BasicProducerSample {
  public static void main(String[] args) throws InterruptedException {
    final var topic = "getting-started";

    final Map<String, Object> config = 
        Map.of(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092", 
               ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName(), 
               ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName(), 
               ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

    try (var producer = new KafkaProducer<String, String>(config)) {
      while (true) {
        final var key = "myKey";
        final var value = new Date().toString();
        out.format("Publishing record with value %s%n", 
                   value);

        final Callback callback = (metadata, exception) -> {
          out.format("Published with metadata: %s, error: %s%n", 
                     metadata, exception);
        };

        // publish the record, handling the metadata in the callback
        producer.send(new ProducerRecord<>(topic, key, value), callback);

        // wait a second before publishing another
        Thread.sleep(1000);
      }
    }
  }
}

      /** A sample Kafka consumer. */

import static java.lang.System.*;

import java.time.*;
import java.util.*;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.*;

public final class BasicConsumerSample {
  public static void main(String[] args) {
    final var topic = "getting-started";

    final Map<String, Object> config = 
        Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092", 
               ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName(), 
               ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName(), 
               ConsumerConfig.GROUP_ID_CONFIG, "basic-consumer-sample",
               ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest",
               ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

    try (var consumer = new KafkaConsumer<String, String>(config)) {
      consumer.subscribe(Set.of(topic));

      while (true) {
        final var records = consumer.poll(Duration.ofMillis(100));
        for (var record : records) {
          out.format("Got record with value %s%n", record.value());
        }
        consumer.commitAsync();
      }
    }
  }
}

Очевидно, что это не модульные тесты. Но с очень небольшой доработкой их можно было превратить в одно целое. Следующим шагом будет удаление Thread.sleep()и добавить утверждения. Обратите внимание: поскольку Kafka по своей природе асинхронна, наивное утверждение опубликованного сообщения в потребителе сразу после публикации не удастся. Для надежного воспроизводимого теста вы можете использовать что-то вроде Timesert.