Не за горами выход релиза 4.0 Apache Kafka. Согласно графику релиза, 15 января состоится code freeze, а через пару недель или позже, после стабилизации, версия 4.0 увидит свет. Самое время присмотреться, что же в неё вошло.

Развитие Apache Kafka происходит в рамках процесса процесса работы с KIP'ами, Kafka Improvement Proposals. В релиз 4.0 их вошло 37 штук. Из них наиболее интересной мне показалась доработка, связанная с введением новой концепции, которую разработчики смело сравнили с очередями. Давайте посмотрим, что у них получилось. Остальные доработки, наверное, важны не менее.

Очереди!

"Очередям" в релизе посвящено 2 KIP'а:

Как обычно, новая функциональность заявлена как экспериментальная, изменяющаяся и не рекомендуется к применению в продуктиве. Использование нового класса, в целом, аналогично старому.

Поддерживаются различные сценарии подтверждения, неявный:

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props);
consumer.subscribe(Arrays. asList("foo"));
while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration. ofMillis(100));
  for (ConsumerRecord<String, String> record : records) {
    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    doProcessing(record);
  }
}  

Альтернативно, можно использовать consumer.commitSync() или consumer.commitAsync() для передачи подтверждений, но это менее эффективно, поскольку приводит к дополнительным вызовам. Возможно также подтверждение на уровне отдельных записей:

Properties props = new Properties();
props.setProperty("bootstrap. servers", "localhost:9092");
props.setProperty("group.id", "test");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props);
consumer.subscribe(Arrays. asList("foo"));
while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  for (ConsumerRecord<String, String> record : records) {
    try {
      doProcessing(record);
      consumer.acknowledge(record, AcknowledgeType.ACCEPT);
    } catch (Exception e) {
      consumer.acknowledge(record, AcknowledgeType.REJECT);
      break;
    }
  }
  consumer.commitSync();
}

Транзакционное поведение групп определяется свойством group.share.isolation.level. Оно применяется ко всей группе, а не к отдельному потребителю. Свойство может принимать значение read_committed или read_uncommitted (по-умолчанию).

В рамках данного KIP также добавлена новая утилита командной строки kafka-console-share-consumer.sh. Помимо вышеперечисленных, добавлено множество новых свойств, изменено несколько старых, появились новые API. К сожалению, новой -perf.sh утилиты пока нет.

В общем, это новая, обширная и интересная тема, которая заслуживает подробного разбора.

Что ещё интересного?

Вот ещё несколько улучшений релиза:

KIP-724 удалил совместимость с клиентами и брокерами версии 2.1 и ниже. Обновляйтесь!

Также в релизе в модулях broker и tools удалена поддержка Java 11.

Важными с точки зрения совместимости кода являются обновления библиотек:

• KIP-1032

  • jetty 9.4.56 → 12.0.15

  • javax.ws.rs → Jakarta RESTful WS

• KIP-653

  • log4j → log4j2