add CHANGELOG and fix nits

Author: antonio-pedro99

CHANGELOG.md

@@ -3,6 +3,9 @@
 ## 0.30.0
 
 * Dependency updates (Prometheus JMX Collector 1.0.1, Prometheus Client 1.3.1)
+* Added support for message timestamp.
+  1. Implemented support for interpreting the timestamp parameter in `ProducerRecord`objects sent to Kafka topics via the bridge.
+  2. Allow users to read the `ConsumerRecord`'s timestamp on the request's response.
 
 ## 0.29.0
 

src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java

@@ -50,7 +50,7 @@ public ProducerRecord<byte[], byte[]> toKafkaRecord(String kafkaTopic, Integer p
             }
             if (json.has("headers")) {
                 ArrayNode jsonArray = (ArrayNode) json.get("headers");
-                for (JsonNode jsonObject : jsonArray) {
+                for (JsonNode jsonObject: jsonArray) {
                     headers.add(new RecordHeader(jsonObject.get("key").asText(), DatatypeConverter.parseBase64Binary(jsonObject.get("value").asText())));
                 }
             }
@@ -106,7 +106,7 @@ public byte[] toMessages(ConsumerRecords<byte[], byte[]> records) {
 
             ArrayNode headers = JsonUtils.createArrayNode();
 
-            for (Header kafkaHeader : record.headers()) {
+            for (Header kafkaHeader: record.headers()) {
                 ObjectNode header = JsonUtils.createObjectNode();
 
                 header.put("key", kafkaHeader.key());

src/test/java/io/strimzi/kafka/bridge/clients/BasicKafkaClient.java

@@ -61,10 +61,10 @@ public int sendStringMessagesPlain(long timeoutMs, String topicName, int message
         properties.setProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.PLAINTEXT.name);
 
         try (Producer plainProducer = new Producer.ProducerBuilder(resultPromise, msgCntPredicate, topicName, message, partition, timestamp)
-                .withProperties(properties)
-                .withHeaders(headers)
-                .withNullKeyRecord(withNullKeyRecord)
-                .build()) {
+           .withProperties(properties)
+           .withHeaders(headers)
+           .withNullKeyRecord(withNullKeyRecord)
+           .build()) {
 
             plainProducer.getVertx().deployVerticle(plainProducer);
 
@@ -84,7 +84,7 @@ public int sendStringMessagesPlain(long timeoutMs, String topicName, int message
      */
     public int sendStringMessagesPlain(String topicName, int messageCount) {
         return sendStringMessagesPlain(Duration.ofMinutes(2).toMillis(), topicName, messageCount,
-                List.of(), "\"Hello\" : \"World\"", 0, null, false);
+            List.of(), "\"Hello\" : \"World\"", 0, null, false);
     }
 
     /**
@@ -93,17 +93,17 @@ public int sendStringMessagesPlain(String topicName, int messageCount) {
      * @param topicName    topic name where messages are send
      * @param message      content to be sent
      * @param messageCount message count
-     * @param partition    partition, which will be selected
+     * @param partition partition, which will be selected
      * @return sent message count
      */
     public int sendStringMessagesPlain(String topicName, String message, int messageCount, int partition) {
         return sendStringMessagesPlain(Duration.ofMinutes(2).toMillis(), topicName, messageCount,
-                List.of(), message, partition, null, true);
+            List.of(), message, partition, null, true);
     }
 
     public int sendStringMessagesPlain(String topicName, String message, int messageCount, int partition, boolean withNullKeyRecord) {
         return sendStringMessagesPlain(Duration.ofMinutes(2).toMillis(), topicName, messageCount,
-                List.of(), message, partition, null, withNullKeyRecord);
+            List.of(), message, partition, null, withNullKeyRecord);
     }
 
     /**
@@ -133,10 +133,10 @@ public int sendJsonMessagesPlain(long timeoutMs, String topicName, int messageCo
         properties.setProperty(ProducerConfig.CLIENT_ID_CONFIG, "producer-sender-plain-" + new Random().nextInt(Integer.MAX_VALUE));
         properties.setProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.PLAINTEXT.name);
         try (Producer plainProducer = new Producer.ProducerBuilder(resultPromise, msgCntPredicate, topicName, message, partition, timestamp)
-                .withProperties(properties)
-                .withHeaders(headers)
-                .withNullKeyRecord(withNullKeyRecord)
-                .build()) {
+            .withProperties(properties)
+            .withHeaders(headers)
+            .withNullKeyRecord(withNullKeyRecord)
+            .build()) {
             plainProducer.getVertx().deployVerticle(plainProducer);
 
             return plainProducer.getResultPromise().get(timeoutMs, TimeUnit.MILLISECONDS);
@@ -157,7 +157,7 @@ public int sendJsonMessagesPlain(long timeoutMs, String topicName, int messageCo
      */
     public int sendJsonMessagesPlain(String topicName, int messageCount, String message, Long timestamp) {
         return sendJsonMessagesPlain(Duration.ofMinutes(2).toMillis(), topicName, messageCount, List.of(),
-                message, 0, timestamp,  false);
+                message, 0, timestamp, false);
     }
 
     /**
@@ -172,7 +172,7 @@ public int sendJsonMessagesPlain(String topicName, int messageCount, String mess
      */
     public int sendJsonMessagesPlain(String topicName, int messageCount, String message, int partition, boolean withNullKeyRecord) {
         return sendJsonMessagesPlain(Duration.ofMinutes(2).toMillis(), topicName, messageCount, List.of(),
-                message, partition, null,  withNullKeyRecord);
+            message, partition, null, withNullKeyRecord);
     }
 
     /**
@@ -195,26 +195,26 @@ public int sendJsonMessagesPlain(String topicName, int messageCount, List<KafkaH
      *
      * @param topicName    topic name where messages are send
      * @param messageCount message count
-     * @param message      specific message to send
-     * @param partition    partition count, how many shards/partitions will topic have
+     * @param message specific message to send
+     * @param partition partition count, how many shards/partitions will topic have
      * @return sent message count
      */
     public int sendJsonMessagesPlain(String topicName, int messageCount, String message, int partition) {
         return sendJsonMessagesPlain(Duration.ofMinutes(2).toMillis(), topicName, messageCount, List.of(),
-                message, partition, null, false);
+            message, partition, null, false);
     }
 
     /**
      * Send messages to entry-point of the kafka cluster with PLAINTEXT security protocol setting
      *
      * @param topicName    topic name where messages are send
      * @param messageCount message count
-     * @param message      specific message to send
+     * @param message specific message to send
      * @return sent message count
      */
     public int sendJsonMessagesPlain(String topicName, int messageCount, String message) {
         return sendJsonMessagesPlain(Duration.ofMinutes(2).toMillis(), topicName, messageCount, List.of(),
-                message, 0, null, false);
+            message, 0, null, false);
     }
 
     /**
@@ -226,7 +226,7 @@ public int sendJsonMessagesPlain(String topicName, int messageCount, String mess
      */
     public int sendJsonMessagesPlain(String topicName, int messageCount) {
         return sendJsonMessagesPlain(Duration.ofMinutes(2).toMillis(), topicName, messageCount, List.of(),
-                "{\"Hello\" : \"World\"}", 0, null, false);
+            "{\"Hello\" : \"World\"}", 0, null, false);
     }
 
     /**