Merge pull request #171 from companieshouse/feature/JU-89-java-21-migration

custom kafka producer implementation

Author: foman-ch

pom.xml

@@ -61,17 +61,17 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
+            <dependency>
+                <groupId>org.testcontainers</groupId>
+                <artifactId>testcontainers-bom</artifactId>
+                <version>${test-containers.version}</version>
+                <scope>import</scope>
+                <type>pom</type>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
     <dependencies>
-        <dependency>
-            <groupId>org.testcontainers</groupId>
-            <artifactId>testcontainers-bom</artifactId>
-            <version>${test-containers.version}</version>
-            <type>pom</type>
-            <scope>import</scope>
-        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>

src/main/java/uk/gov/companieshouse/orders/api/kafka/OrdersApiKafkaProducer.java

@@ -0,0 +1,98 @@
+package uk.gov.companieshouse.orders.api.kafka;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import uk.gov.companieshouse.kafka.message.Message;
+import uk.gov.companieshouse.kafka.producer.ProducerConfig;
+import uk.gov.companieshouse.kafka.producer.factory.KafkaProducerFactory;
+
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+public class OrdersApiKafkaProducer {
+
+    private KafkaProducer<String, byte[]> kafkaProducer;
+
+    /**
+     * Instantiate the necessary configuration for producing and a producer for Kafka.
+     * The {@link KafkaProducerFactory} will be used to create the producer.
+     *
+     * @param config Configuration for the producer
+     */
+    public OrdersApiKafkaProducer(ProducerConfig config) {
+        this(config, new KafkaProducerFactory());
+    }
+
+    /**
+     * Instantiate the necessary configuration for producing and a producer for Kafka.
+     *
+     * @param config Configuration for the producer
+     * @param producerFactory Factory to create the producer
+     */
+    public OrdersApiKafkaProducer(ProducerConfig config, KafkaProducerFactory producerFactory) {
+        Properties props = new Properties();
+
+        props.put("bootstrap.servers", String.join(",", config.getBrokerAddresses()));
+        props.put("acks", config.getAcks().getCode());
+        props.put("key.serializer", config.getKeySerializer());
+        props.put("value.serializer", config.getValueSerializer());
+        props.put("retries", config.getRetries());
+        props.put("max.block.ms", config.getMaxBlockMilliseconds());
+        props.put("request.timeout.ms", config.getRequestTimeoutMilliseconds());
+        props.put("enable.idempotence", false);
+
+        if (config.isRoundRobinPartitioner()) {
+            props.put("partition.assignment.strategy", "roundrobin");
+        }
+
+        kafkaProducer = producerFactory.getProducer(props);
+    }
+
+    /**
+     * Send a message to a topic in Kafka.
+     *
+     * The data in the message is sent to Kafka in a byte array
+     * to avoid unrecognised encoding characters which can occur
+     * when Strings are used with Kafka.
+     *
+     */
+    public void send(Message msg) throws ExecutionException, InterruptedException {
+
+        ProducerRecord<String, byte[]> record = getProducerRecordFromMessage(msg);
+
+        kafkaProducer.send(record).get();
+    }
+
+    /**
+     * Send a message to a topic in Kafka, and return a {@link Future < RecordMetadata >} for processing manually.
+     *
+     * The data in the message is sent to Kafka in a byte array
+     * to avoid unrecognised encoding characters which can occur
+     * when Strings are used with Kafka.
+     *
+     */
+    public Future<RecordMetadata> sendAndReturnFuture(Message msg) {
+
+        ProducerRecord<String, byte[]> record = getProducerRecordFromMessage(msg);
+
+        return kafkaProducer.send(record);
+    }
+
+    private ProducerRecord<String, byte[]> getProducerRecordFromMessage(Message msg) {
+
+        return new ProducerRecord<>(
+                msg.getTopic(),
+                msg.getPartition(),
+                msg.getTimestamp(),
+                msg.getKey(),
+                msg.getValue()
+        );
+    }
+
+    public void close() {
+        kafkaProducer.close();
+    }
+
+}

src/main/java/uk/gov/companieshouse/orders/api/kafka/OrdersKafkaProducer.java

@@ -24,7 +24,7 @@
 @Service
 public class OrdersKafkaProducer implements InitializingBean {
     private static final Logger LOGGER = LoggerFactory.getLogger(APPLICATION_NAME_SPACE);
-    private CHKafkaProducer chKafkaProducer;
+    private OrdersApiKafkaProducer chKafkaProducer;
     @Value("${spring.kafka.producer.bootstrap-servers}")
     private String brokerAddresses;
 
@@ -61,6 +61,6 @@ public void afterPropertiesSet() {
         config.setAcks(Acks.WAIT_FOR_ALL);
         config.setRetries(10);
         config.setMaxBlockMilliseconds(10000);
-        chKafkaProducer = new CHKafkaProducer(config);
+        chKafkaProducer = new OrdersApiKafkaProducer(config);
     }
 }

src/main/resources/application.properties

@@ -16,7 +16,6 @@ spring.data.mongodb.uri=${MONGODB_URL}
 spring.data.mongodb.field-naming-strategy=uk.gov.companieshouse.orders.api.model.NoIsSnakeCaseFieldNamingStrategy
 
 spring.kafka.producer.bootstrap-servers: ${KAFKA_BROKER_ADDR}
-spring.kafka.producer.properties.enable.idempotence=false
 
 feature.options.ordersSearchEndpointEnabled=${ORDERS_SEARCH_ENDPOINT_ENABLED:false}
 feature.options.multiItemBasketSearchEnabled=${ORDERS_SEARCH_MULTIBASKET_ENABLED:false}