Upgrade Kafka Client

[cer]

The 2.3.0 client is very old. It should be replaced with 3.x

eventuate-messaging-kafka/eventuate-messaging-kafka-common/build.gradle

Line 5 in 32951a1

compile 'org.apache.kafka:kafka-clients:2.3.0'

We should consider upgrading the broker too:

eventuate-messaging-kafka/kafka/Dockerfile

Line 2 in 389ee45

ENV KAFKA_VERSION=kafka_2.13-2.8.1

[ddaeyongee]

@cer
Hi. This is Taeyong from LG organization.

Currently, In our project, we do not use the apache kafka client license directly, but purchase a paid license for the confluent platform and use it.
The confluent platform version used is 6.1, and the confluent(kafka) version 6.1.1 is compatible with apache kafka 2.7.x.
According to the "Confluent Platform 3.2 and later" section of confluent's guide,

 "Kafka Java Clients that are included in Confluent Platform 3.2 (Kafka version 0.10.2) and later are compatible with any Kafka broker that is included in Confluent Platform 3.0 and later. " 

link : https://docs.confluent.io/platform/current/installation/versions-interoperability.html#ak-java-clients

So, compatibility between kafka client 3.x and confluent(kafka) 6.1.1 seems to be fine.
Thank you for your active support.

[cer]

Also, need to review/optimize:

• Consumer properties -

  eventuate-messaging-kafka/eventuate-messaging-kafka-basic-consumer/src/main/java/io/eventuate/messaging/kafka/basic/consumer/ConsumerPropertiesFactory.java

  Lines 7 to 16 in 5cef97b

  	Properties consumerProperties = new Properties();
  	consumerProperties.put("bootstrap.servers", bootstrapServers);
  	consumerProperties.put("group.id", subscriberId);
  	consumerProperties.put("enable.auto.commit", "false");
  	//consumerProperties.put("auto.commit.interval.ms", "1000");
  	consumerProperties.put("session.timeout.ms", "30000");
  	consumerProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  	consumerProperties.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
  	consumerProperties.put("auto.offset.reset", "earliest");
  	return consumerProperties;

• Producer properties -

  eventuate-messaging-kafka/eventuate-messaging-kafka-producer/src/main/java/io/eventuate/messaging/kafka/producer/EventuateKafkaProducer.java

  Lines 25 to 34 in 5cef97b

  	producerProps = new Properties();
  	producerProps.put("bootstrap.servers", bootstrapServers);
  	producerProps.put("acks", "all");
  	producerProps.put("retries", 0);
  	producerProps.put("batch.size", 16384);
  	producerProps.put("linger.ms", 1);
  	producerProps.put("buffer.memory", 33554432);
  	producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  	producerProps.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
  	producerProps.putAll(eventuateKafkaProducerConfigurationProperties.getProperties());

For example, enable idempotent producer + ack=all + etc.

[cer]

Notes on version compatibility:

• Bidirectional version compatibility - https://www.confluent.io/blog/upgrading-apache-kafka-clients-just-got-easier/
  • Previously: client could not be newer than broker
  • Now: new clients can talk to older brokers
  • An exception is thrown if broker does not support an operation

[cer]

Kafka Client 3.1.1 seems like a reasonable choice - used by Spring Boot dependencies.