Создаем Spring Boot Starter для Kafka с Avro: пошаговое руководство

package com.app.petr;  import org.springframework.boot.context.properties.ConfigurationProperties;  import java.util.HashMap; import java.util.Map;  @ConfigurationProperties(prefix = "apppetr.kafka") public class KafkaProperties {     private String bootstrapServers = "localhost:9092";     private Producer producer = new Producer();     private Consumer consumer = new Consumer();     private String avroSchemaPath;       public static class Producer {         private String topic = "default-topic";         private boolean idempotenceEnabled = true; // Идемпотентность для надежности         private int acks = 1; // Подтверждения: 0, 1, all         private int retries = 3; // Количество попыток при сбоях         private int retryBackoffMs = 1000; // Задержка между ретраями         private int deliveryTimeoutMs = 120000; // Таймаут доставки         private int requestTimeoutMs = 30000; // Таймаут запроса         private int maxInFlightRequests = 5; // Макс. количество неподтвержденных запросов         private Map<String, Object> config = new HashMap<>(); // Дополнительные настройки          public String getTopic() { return topic; }         public void setTopic(String topic) { this.topic = topic; }         public boolean isIdempotenceEnabled() { return idempotenceEnabled; }         public void setIdempotenceEnabled(boolean idempotenceEnabled) { this.idempotenceEnabled = idempotenceEnabled; }         public int getAcks() { return acks; }         public void setAcks(int acks) { this.acks = acks; }         public int getRetries() { return retries; }         public void setRetries(int retries) { this.retries = retries; }         public int getRetryBackoffMs() { return retryBackoffMs; }         public void setRetryBackoffMs(int retryBackoffMs) { this.retryBackoffMs = retryBackoffMs; }         public int getDeliveryTimeoutMs() { return deliveryTimeoutMs; }         public void setDeliveryTimeoutMs(int deliveryTimeoutMs) { this.deliveryTimeoutMs = deliveryTimeoutMs; }         public int getRequestTimeoutMs() { return requestTimeoutMs; }         public void setRequestTimeoutMs(int requestTimeoutMs) { this.requestTimeoutMs = requestTimeoutMs; }         public int getMaxInFlightRequests() { return maxInFlightRequests; }         public void setMaxInFlightRequests(int maxInFlightRequests) { this.maxInFlightRequests = maxInFlightRequests; }         public Map<String, Object> getConfig() { return config; }         public void setConfig(Map<String, Object> config) { this.config = config; }     }      public static class Consumer {         private String topic = "default-topic";         private String groupId = "default-group";         private int maxPollRecords = 500; // Макс. записей за один poll         private int maxPollIntervalMs = 300000; // Макс. интервал между poll         private int sessionTimeoutMs = 10000; // Таймаут сессии         private int heartbeatIntervalMs = 3000; // Интервал heartbeat         private int fetchMaxBytes = 52428800; // Макс. размер выборки (50MB)         private boolean autoCommitEnabled = true; // Автокоммит оффсетов         private int autoCommitIntervalMs = 5000; // Интервал автокоммита         private String autoOffsetReset = "earliest"; // Сброс оффсета         private int retryBackoffMs = 1000; // Задержка между ретраями         private int maxRetries = 3; // Количество ретраев         private Map<String, Object> config = new HashMap<>(); // Дополнительные настройки          public String getTopic() { return topic; }         public void setTopic(String topic) { this.topic = topic; }         public String getGroupId() { return groupId; }         public void setGroupId(String groupId) { this.groupId = groupId; }         public int getMaxPollRecords() { return maxPollRecords; }         public void setMaxPollRecords(int maxPollRecords) { this.maxPollRecords = maxPollRecords; }         public int getMaxPollIntervalMs() { return maxPollIntervalMs; }         public void setMaxPollIntervalMs(int maxPollIntervalMs) { this.maxPollIntervalMs = maxPollIntervalMs; }         public int getSessionTimeoutMs() { return sessionTimeoutMs; }         public void setSessionTimeoutMs(int sessionTimeoutMs) { this.sessionTimeoutMs = sessionTimeoutMs; }         public int getHeartbeatIntervalMs() { return heartbeatIntervalMs; }         public void setHeartbeatIntervalMs(int heartbeatIntervalMs) { this.heartbeatIntervalMs = heartbeatIntervalMs; }         public int getFetchMaxBytes() { return fetchMaxBytes; }         public void setFetchMaxBytes(int fetchMaxBytes) { this.fetchMaxBytes = fetchMaxBytes; }         public boolean isAutoCommitEnabled() { return autoCommitEnabled; }         public void setAutoCommitEnabled(boolean autoCommitEnabled) { this.autoCommitEnabled = autoCommitEnabled; }         public int getAutoCommitIntervalMs() { return autoCommitIntervalMs; }         public void setAutoCommitIntervalMs(int autoCommitIntervalMs) { this.autoCommitIntervalMs = autoCommitIntervalMs; }         public String getAutoOffsetReset() { return autoOffsetReset; }         public void setAutoOffsetReset(String autoOffsetReset) { this.autoOffsetReset = autoOffsetReset; }         public int getRetryBackoffMs() { return retryBackoffMs; }         public void setRetryBackoffMs(int retryBackoffMs) { this.retryBackoffMs = retryBackoffMs; }         public int getMaxRetries() { return maxRetries; }         public void setMaxRetries(int maxRetries) { this.maxRetries = maxRetries; }         public Map<String, Object> getConfig() { return config; }         public void setConfig(Map<String, Object> config) { this.config = config; }     }      public String getBootstrapServers() { return bootstrapServers; }     public void setBootstrapServers(String bootstrapServers) { this.bootstrapServers = bootstrapServers; }     public Producer getProducer() { return producer; }     public void setProducer(Producer producer) { this.producer = producer; }     public Consumer getConsumer() { return consumer; }     public void setConsumer(Consumer consumer) { this.consumer = consumer; }      public String getAvroSchemaPath() {         return avroSchemaPath;     }      public void setAvroSchemaPath(String avroSchemaPath) {         this.avroSchemaPath = avroSchemaPath;     }  }

package com.app.petr;  import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.ProducerConfig; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.EnableKafka; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.*; import org.springframework.kafka.listener.DefaultErrorHandler; import org.springframework.util.backoff.BackOff; import org.springframework.util.backoff.FixedBackOff;  import java.util.HashMap; import java.util.Map;  @Configuration @ConditionalOnClass({ KafkaTemplate.class, ConcurrentKafkaListenerContainerFactory.class }) @EnableKafka @EnableConfigurationProperties(KafkaProperties.class) @ConditionalOnProperty(prefix = "apppetr.kafka", name = "bootstrap-servers") public class KafkaAutoConfiguration {      private final KafkaProperties properties;      public KafkaAutoConfiguration(KafkaProperties properties) {         this.properties = properties;     }      // Producer Configuration     @Bean     @ConditionalOnMissingBean     public ProducerFactory<String, byte[]> producerFactory() {         Map<String, Object> config = new HashMap<>();         config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());         config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");         config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");         config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, properties.getProducer().isIdempotenceEnabled());         config.put(ProducerConfig.ACKS_CONFIG, String.valueOf(properties.getProducer().getAcks()));         config.put(ProducerConfig.RETRIES_CONFIG, properties.getProducer().getRetries());         config.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, properties.getProducer().getRetryBackoffMs());         config.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, properties.getProducer().getDeliveryTimeoutMs());         config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, properties.getProducer().getRequestTimeoutMs());         config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, properties.getProducer().getMaxInFlightRequests());         config.putAll(properties.getProducer().getConfig());         return new DefaultKafkaProducerFactory<>(config);     }      @Bean     @ConditionalOnMissingBean     public KafkaTemplate<String, byte[]> kafkaTemplate(ProducerFactory<String, byte[]> producerFactory) {         KafkaTemplate<String, byte[]> template = new KafkaTemplate<>(producerFactory);         template.setDefaultTopic(properties.getProducer().getTopic());         return template;     }      // Consumer Configuration     @Bean     @ConditionalOnMissingBean     public ConsumerFactory<String, byte[]> consumerFactory() {         Map<String, Object> config = new HashMap<>();         config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());         config.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getConsumer().getGroupId());         config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");         config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");         config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, properties.getConsumer().getMaxPollRecords());         config.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, properties.getConsumer().getMaxPollIntervalMs());         config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, properties.getConsumer().getSessionTimeoutMs());         config.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, properties.getConsumer().getHeartbeatIntervalMs());         config.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, properties.getConsumer().getFetchMaxBytes());         config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, properties.getConsumer().isAutoCommitEnabled());         config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, properties.getConsumer().getAutoCommitIntervalMs());         config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, properties.getConsumer().getAutoOffsetReset());         config.putAll(properties.getConsumer().getConfig());         return new DefaultKafkaConsumerFactory<>(config);     }      @Bean     @ConditionalOnMissingBean     public ConcurrentKafkaListenerContainerFactory<String, byte[]> kafkaListenerContainerFactory(             ConsumerFactory<String, byte[]> consumerFactory) {         ConcurrentKafkaListenerContainerFactory<String, byte[]> factory =                 new ConcurrentKafkaListenerContainerFactory<>();         factory.setConsumerFactory(consumerFactory);         factory.setConcurrency(1); // Параллелизм         factory.getContainerProperties().setPollTimeout(3000); // Таймаут опроса         factory.setCommonErrorHandler(errorHandler()); // Настраиваем обработку ошибок с ретраями         return factory;     }      @Bean     public DefaultErrorHandler errorHandler() {         BackOff backOff = new FixedBackOff(                 properties.getConsumer().getRetryBackoffMs(),                 properties.getConsumer().getMaxRetries()         );         DefaultErrorHandler errorHandler = new DefaultErrorHandler(                 (record, exception) -> {                     System.err.println("Failed to process record: " + record + ", exception: " + exception.getMessage());                 }, // Логирование ошибок                 backOff         );         return errorHandler;     } }

package com.app.petr;  import app.petr.Message; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.KafkaTemplate;  import java.nio.ByteBuffer;  @SpringBootApplication public class ProducerApplication {      public static void main(String[] args) {         SpringApplication.run(ProducerApplication.class, args);     }      @Bean     public ApplicationRunner runner(KafkaTemplate<String, byte[]> kafkaTemplate, KafkaProperties properties) {         return args -> {             String topic = properties.getProducer().getTopic();             if (topic == null) {                 throw new IllegalStateException("Producer topic is not configured");             }              for (int i = 0; i < 100; i++) {                 Message message = Message.newBuilder()                         .setId("id-" + i)                         .setContent("Message " + i)                         .setTimestamp(System.currentTimeMillis())                         .build();                  ByteBuffer buffer = message.toByteBuffer();                 byte[] messageBytes = new byte[buffer.remaining()];                 buffer.get(messageBytes);                  kafkaTemplate.send(topic, message.getId().toString(), messageBytes);                 System.out.println("Sent message: " + message.getId());                 Thread.sleep(1000);             }         };     } }

package com.app.petr;   import app.petr.Message; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.kafka.annotation.KafkaListener;  import java.nio.ByteBuffer;  @SpringBootApplication public class ConsumerApplication {      public static void main(String[] args) {         SpringApplication.run(ConsumerApplication.class, args);     }      @KafkaListener(topics = "#{kafkaProperties.consumer.topic}", groupId = "#{kafkaProperties.consumer.groupId}")     public void listen(byte[] message) throws Exception {         Message avroMessage = Message.fromByteBuffer(ByteBuffer.wrap(message));         System.out.printf("Received message: id=%s, content=%s, timestamp=%d%n",                 avroMessage.getId(), avroMessage.getContent(), avroMessage.getTimestamp());     } }

package com.app.petr;  import app.petr.Message; import org.apache.avro.io.Decoder; import org.apache.avro.io.DecoderFactory; import org.apache.avro.io.Encoder; import org.apache.avro.io.EncoderFactory; import org.apache.avro.specific.SpecificDatumReader; import org.apache.avro.specific.SpecificDatumWriter; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.junit.jupiter.api.Test; import org.testcontainers.containers.KafkaContainer; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.utility.DockerImageName;  import java.io.ByteArrayOutputStream; import java.time.Duration; import java.util.Collections; import java.util.Properties;  import static org.assertj.core.api.Assertions.assertThat;  @Testcontainers public class IntegrationProducerAndConsumerTest {      private static final String TOPIC = "rest_data";      @Container     private static final KafkaContainer kafkaContainer = new KafkaContainer(             DockerImageName.parse("confluentinc/cp-kafka:7.3.0"));      private KafkaProducer<String, byte[]> createProducer() {         Properties props = new Properties();         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");         return new KafkaProducer<>(props);     }      private KafkaConsumer<String, byte[]> createConsumer() {         Properties props = new Properties();         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());         props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");         return new KafkaConsumer<>(props);     }      @Test     public void testProducerAndConsumer() throws Exception {          KafkaProducer<String, byte[]> producer = createProducer();         KafkaConsumer<String, byte[]> consumer = createConsumer();          consumer.subscribe(Collections.singleton(TOPIC));          Message message = Message.newBuilder()                 .setId("test-id")                 .setContent("Test content")                 .setTimestamp(System.currentTimeMillis())                 .build();          ByteArrayOutputStream out = new ByteArrayOutputStream();         Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);         SpecificDatumWriter<Message> writer = new SpecificDatumWriter<>(Message.class);         writer.write(message, encoder);         encoder.flush();         byte[] serializedMessage = out.toByteArray();          ProducerRecord<String, byte[]> record = new ProducerRecord<>(TOPIC, message.getId().toString(), serializedMessage);         producer.send(record).get();         producer.flush();          SpecificDatumReader<Message> reader = new SpecificDatumReader<>(Message.class);         ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(5));         assertThat(records.isEmpty()).isFalse();          records.forEach(consumerRecord -> {             try {                 Decoder decoder = DecoderFactory.get().binaryDecoder(consumerRecord.value(), null);                 Message receivedMessage = reader.read(null, decoder);                 assertThat(receivedMessage.getId().toString()).isEqualTo(message.getId());                 assertThat(receivedMessage.getContent().toString()).isEqualTo(message.getContent());                 assertThat(receivedMessage.getTimestamp()).isEqualTo(message.getTimestamp());             } catch (Exception e) {                 throw new RuntimeException("Failed to deserialize message", e);             }         });          producer.close();         consumer.close();     } }

package com.app.petr;  import app.petr.Message; import org.apache.avro.io.Encoder; import org.apache.avro.io.EncoderFactory; import org.apache.avro.specific.SpecificDatumWriter; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.errors.TimeoutException; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.testcontainers.containers.KafkaContainer; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.utility.DockerImageName;  import java.io.ByteArrayOutputStream; import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future;  import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows;  @Testcontainers public class ProducerApplicationTest {      private static final String TOPIC = "test_data";      @Container     private static final KafkaContainer kafkaContainer = new KafkaContainer(             DockerImageName.parse("confluentinc/cp-kafka:7.3.0"));      @BeforeAll     static void setup() {         kafkaContainer.start();     }      private KafkaProducer<String, byte[]> createProducer(Properties props) {         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");         return new KafkaProducer<>(props);     }      private byte[] serializeMessage(Message message) throws Exception {         ByteArrayOutputStream out = new ByteArrayOutputStream();         Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);         SpecificDatumWriter<Message> writer = new SpecificDatumWriter<>(Message.class);         writer.write(message, encoder);         encoder.flush();         return out.toByteArray();     }      @Test     public void testProducerWithDefaultConfig() throws Exception {         Properties props = new Properties();         KafkaProducer<String, byte[]> producer = createProducer(props);          Message message = Message.newBuilder()                 .setId("test-id-1")                 .setContent("Default config test")                 .setTimestamp(System.currentTimeMillis())                 .build();          byte[] serializedMessage = serializeMessage(message);         ProducerRecord<String, byte[]> record = new ProducerRecord<>(TOPIC, message.getId().toString(), serializedMessage);          Future<?> future = producer.send(record);         producer.flush();          assertThat(future.get()).isNotNull(); // Проверяем успешную отправку         producer.close();     }      @Test     public void testProducerWithIdempotenceEnabled() throws Exception {         Properties props = new Properties();         props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); // Включаем идемпотентность         props.put(ProducerConfig.ACKS_CONFIG, "all"); // Требуется для идемпотентности         props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5"); // Ограничение для идемпотентности         KafkaProducer<String, byte[]> producer = createProducer(props);          Message message = Message.newBuilder()                 .setId("test-id-2")                 .setContent("Idempotent test")                 .setTimestamp(System.currentTimeMillis())                 .build();          byte[] serializedMessage = serializeMessage(message);         ProducerRecord<String, byte[]> record = new ProducerRecord<>(TOPIC, message.getId().toString(), serializedMessage);          Future<?> future = producer.send(record);         producer.send(record); // Отправляем тот же ключ ещё раз         producer.flush();          assertThat(future.get()).isNotNull(); // Успешная отправка с идемпотентностью         producer.close();     }      @Test     public void testProducerWithAcksZero() throws Exception {         Properties props = new Properties();         props.put(ProducerConfig.ACKS_CONFIG, "0"); // Без подтверждений         KafkaProducer<String, byte[]> producer = createProducer(props);          Message message = Message.newBuilder()                 .setId("test-id-3")                 .setContent("Acks=0 test")                 .setTimestamp(System.currentTimeMillis())                 .build();          byte[] serializedMessage = serializeMessage(message);         ProducerRecord<String, byte[]> record = new ProducerRecord<>(TOPIC, message.getId().toString(), serializedMessage);          Future<?> future = producer.send(record);         producer.flush();          assertThat(future.get()).isNotNull(); // Отправка без ожидания подтверждения         producer.close();     }      @Test     public void testProducerWithRetries() throws Exception {         Properties props = new Properties();         props.put(ProducerConfig.RETRIES_CONFIG, "3"); // 3 попытки         props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "100"); // Задержка между попытками 100 мс         props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "500"); // Уменьшаем таймаут запроса до 500 мс         props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "1000"); // Таймаут доставки 1000 мс (должен быть >= linger.ms + request.timeout.ms)         props.put(ProducerConfig.LINGER_MS_CONFIG, "0"); // Явно задаём linger.ms для ясности         KafkaProducer<String, byte[]> producer = createProducer(props);          Message message = Message.newBuilder()                 .setId("test-id-4")                 .setContent("Retries test")                 .setTimestamp(System.currentTimeMillis())                 .build();          byte[] serializedMessage = serializeMessage(message);         ProducerRecord<String, byte[]> record = new ProducerRecord<>(TOPIC, message.getId().toString(), serializedMessage);          Future<?> future = producer.send(record);         producer.flush();          assertThat(future.get()).isNotNull(); // Успешная отправка с ретраями         producer.close();     }      @Test     public void testProducerWithShortTimeout() {         Properties props = new Properties();         props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "1"); // Очень короткий таймаут         props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "1"); // Очень короткий таймаут доставки         KafkaProducer<String, byte[]> producer = createProducer(props);          Message message = Message.newBuilder()                 .setId("test-id-5")                 .setContent("Short timeout test")                 .setTimestamp(System.currentTimeMillis())                 .build();          byte[] serializedMessage;         try {             serializedMessage = serializeMessage(message);         } catch (Exception e) {             throw new RuntimeException("Serialization failed", e);         }         ProducerRecord<String, byte[]> record = new ProducerRecord<>(TOPIC, message.getId().toString(), serializedMessage);          Future<?> future = producer.send(record);          // Ожидаем исключение из-за короткого таймаута         assertThrows(ExecutionException.class, future::get, "Expected timeout exception due to short timeout");         assertThat(future.isDone()).isTrue();          producer.close();     } }

package com.app.petr;  import app.petr.Message; import org.apache.avro.io.Decoder; import org.apache.avro.io.DecoderFactory; import org.apache.avro.io.Encoder; import org.apache.avro.io.EncoderFactory; import org.apache.avro.specific.SpecificDatumReader; import org.apache.avro.specific.SpecificDatumWriter; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.testcontainers.containers.KafkaContainer; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.utility.DockerImageName;  import java.io.ByteArrayOutputStream; import java.time.Duration; import java.util.Collections; import java.util.Properties;  import static org.assertj.core.api.Assertions.assertThat;  @Testcontainers public class ConsumerApplicationTest {      private static final String TOPIC = "consumer_test_data";      @Container     private static final KafkaContainer kafkaContainer = new KafkaContainer(             DockerImageName.parse("confluentinc/cp-kafka:7.3.0"));      @BeforeAll     static void setup() {         kafkaContainer.start();     }      @BeforeEach     void clearTopic() {         Properties adminProps = new Properties();         adminProps.put("bootstrap.servers", kafkaContainer.getBootstrapServers());         try (AdminClient adminClient = AdminClient.create(adminProps)) {             adminClient.deleteTopics(Collections.singleton(TOPIC)).all().get();             adminClient.createTopics(Collections.singleton(new NewTopic(TOPIC, 1, (short) 1))).all().get();         } catch (Exception e) {             // Игнорируем ошибки, если топик не существовал         }     }      private KafkaProducer<String, byte[]> createProducer() {         Properties props = new Properties();         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");         return new KafkaProducer<>(props);     }      private KafkaConsumer<String, byte[]> createConsumer(Properties props) {         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");         return new KafkaConsumer<>(props);     }      private byte[] serializeMessage(Message message) throws Exception {         ByteArrayOutputStream out = new ByteArrayOutputStream();         Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);         SpecificDatumWriter<Message> writer = new SpecificDatumWriter<>(Message.class);         writer.write(message, encoder);         encoder.flush();         return out.toByteArray();     }      private Message deserializeMessage(byte[] data) throws Exception {         Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);         SpecificDatumReader<Message> reader = new SpecificDatumReader<>(Message.class);         return reader.read(null, decoder);     }      private void sendMessages(int count) throws Exception {         KafkaProducer<String, byte[]> producer = createProducer();         for (int i = 0; i < count; i++) {             Message message = Message.newBuilder()                     .setId("id-" + i)                     .setContent("Message " + i)                     .setTimestamp(System.currentTimeMillis())                     .build();             byte[] serializedMessage = serializeMessage(message);             producer.send(new ProducerRecord<>(TOPIC, message.getId().toString(), serializedMessage)).get();         }         producer.flush();         producer.close();     }      @Test     public void testConsumerWithDefaultConfig() throws Exception {         Properties props = new Properties();         props.put(ConsumerConfig.GROUP_ID_CONFIG, "default-group-" + System.nanoTime());         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");         KafkaConsumer<String, byte[]> consumer = createConsumer(props);          consumer.subscribe(Collections.singleton(TOPIC));         sendMessages(5);          ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(5));         assertThat(records.count()).isEqualTo(5);          consumer.close();     }      @Test     public void testConsumerWithMaxPollRecords() throws Exception {         Properties props = new Properties();         props.put(ConsumerConfig.GROUP_ID_CONFIG, "max-poll-group-" + System.nanoTime());         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "2");         KafkaConsumer<String, byte[]> consumer = createConsumer(props);          consumer.subscribe(Collections.singleton(TOPIC));         sendMessages(5);          ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(5));         assertThat(records.count()).isLessThanOrEqualTo(2);          consumer.close();     }      @Test     public void testConsumerWithAutoOffsetResetLatest() throws Exception {         Properties props = new Properties();         props.put(ConsumerConfig.GROUP_ID_CONFIG, "latest-group-" + System.nanoTime());         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");         KafkaConsumer<String, byte[]> consumer = createConsumer(props);          // Подписываемся и вызываем poll для завершения регистрации         consumer.subscribe(Collections.singleton(TOPIC));         consumer.poll(Duration.ofSeconds(20)); // Даём Kafka время зарегистрировать консюмера          // Убеждаемся, что до отправки сообщений ничего не читается         ConsumerRecords<String, byte[]> recordsBefore = consumer.poll(Duration.ofSeconds(1));         assertThat(recordsBefore.isEmpty()).isTrue();          // Отправляем 3 сообщения после полной регистрации         sendMessages(3);          // Читаем новые сообщения         ConsumerRecords<String, byte[]> recordsAfter = consumer.poll(Duration.ofSeconds(5));         assertThat(recordsAfter.count()).isEqualTo(3);          consumer.close();     }      @Test     public void testConsumerWithDisableAutoCommit() throws Exception {         Properties props = new Properties();         props.put(ConsumerConfig.GROUP_ID_CONFIG, "no-auto-commit-group-" + System.nanoTime());         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");         KafkaConsumer<String, byte[]> consumer = createConsumer(props);          consumer.subscribe(Collections.singleton(TOPIC));         sendMessages(5);          ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(5));         assertThat(records.count()).isEqualTo(5);          consumer.close();         consumer = createConsumer(props);         consumer.subscribe(Collections.singleton(TOPIC));         ConsumerRecords<String, byte[]> recordsAgain = consumer.poll(Duration.ofSeconds(5));         assertThat(recordsAgain.count()).isEqualTo(5);          consumer.close();     }      @Test     public void testConsumerWithShortSessionTimeout() throws Exception {         Properties props = new Properties();         props.put(ConsumerConfig.GROUP_ID_CONFIG, "short-session-group-" + System.nanoTime());         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "6000");         props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "2000");         KafkaConsumer<String, byte[]> consumer = createConsumer(props);          consumer.subscribe(Collections.singleton(TOPIC));         sendMessages(3);          ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(5));         assertThat(records.count()).isEqualTo(3);          consumer.close();     } }