Merge branch 'main' into feature/reset-connector-offsets

Author: Dugong42

.github/dependabot.yml

@@ -14,6 +14,23 @@ updates:
     - "type/dependencies"
     - "scope/backend"
 
+- package-ecosystem: docker
+  directory: "/api"
+  schedule:
+    interval: weekly
+    time: "10:00"
+    timezone: Europe/London
+  reviewers:
+    - "kafbat/backend"
+  open-pull-requests-limit: 10
+  ignore:
+  - dependency-name: "azul/zulu-openjdk-alpine"
+    # Limit dependabot pull requests to minor Java upgrades
+    update-types: ["version-update:semver-major"]
+  labels:
+    - "type/dependencies"
+    - "scope/backend"
+
 - package-ecosystem: npm
   directory: "/frontend"
   schedule:

api/Dockerfile

@@ -1,4 +1,7 @@
-FROM azul/zulu-openjdk-alpine:17.0.11-jre-headless
+# The tag is ignored when a sha is included but the reason to add it are:  
+# 1. Self Documentation: It is difficult to find out what the expected tag is given a sha alone
+# 2. Helps dependabot during discovery of upgrades
+FROM azul/zulu-openjdk-alpine:17-jre-headless-latest@sha256:af4df00adaec356d092651af50d9e80fd179f96722d267e79acb564aede10fda
 
 RUN apk add --no-cache \
     # snappy codec

api/src/main/java/io/kafbat/ui/controller/MessagesController.java

@@ -118,10 +118,11 @@ public Mono<ResponseEntity<Flux<TopicMessageEventDTO>>> getTopicMessagesV2(Strin
     if (cursor != null) {
       messagesFlux = messagesService.loadMessages(getCluster(clusterName), topicName, cursor);
     } else {
+      var pollingMode = mode == null ? PollingModeDTO.LATEST : mode;
       messagesFlux = messagesService.loadMessages(
           getCluster(clusterName),
           topicName,
-          ConsumerPosition.create(checkNotNull(mode), checkNotNull(topicName), partitions, timestamp, offset),
+          ConsumerPosition.create(pollingMode, checkNotNull(topicName), partitions, timestamp, offset),
           stringFilter,
           smartFilterId,
           limit,

api/src/main/java/io/kafbat/ui/serdes/ProducerRecordCreator.java

@@ -31,7 +31,7 @@ public ProducerRecord<byte[], byte[]> create(String topic,
 
   private Iterable<Header> createHeaders(Map<String, String> clientHeaders) {
     RecordHeaders headers = new RecordHeaders();
-    clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v.getBytes())));
+    clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v == null ? null : v.getBytes())));
     return headers;
   }
 

api/src/test/java/io/kafbat/ui/service/SendAndReadTests.java

@@ -19,6 +19,7 @@
 import io.kafbat.ui.serdes.builtin.StringSerde;
 import io.kafbat.ui.serdes.builtin.sr.SchemaRegistrySerde;
 import java.time.Duration;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -425,6 +426,25 @@ void topicMessageMetadataJson() {
         });
   }
 
+  @Test
+  void headerValueNullPresentTest() {
+    new SendAndReadSpec()
+        .withKeySchema(JSON_SCHEMA)
+        .withValueSchema(JSON_SCHEMA)
+        .withMsgToSend(
+            new CreateTopicMessageDTO()
+                .key(JSON_SCHEMA_RECORD)
+                .keySerde(SchemaRegistrySerde.name())
+                .content(JSON_SCHEMA_RECORD)
+                .valueSerde(SchemaRegistrySerde.name())
+                .headers(Collections.singletonMap("header123", null))
+        )
+        .doAssert(polled -> {
+          assertThat(polled.getHeaders().get("header123")).isNull();
+        });
+  }
+
+
   @Test
   void noKeyAndNoContentPresentTest() {
     new SendAndReadSpec()

documentation/compose/e2e-tests.yaml

@@ -29,7 +29,8 @@ services:
       KAFKA_CLUSTERS_0_KSQLDBSERVER: http://ksqldb:8088
 
   kafka0:
-    image: confluentinc/cp-kafka:7.2.1
+    image: confluentinc/cp-kafka:7.6.0
+    user: "0:0"
     hostname: kafka0
     container_name: kafka0
     healthcheck:
@@ -58,12 +59,10 @@ services:
       KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
       KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
       KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
-    volumes:
-      - ./scripts/update_run.sh:/tmp/update_run.sh
-    command: "bash -c 'if [ ! -f /tmp/update_run.sh ]; then echo \"ERROR: Did you forget the update_run.sh file that came with this docker-compose.yml file?\" && exit 1 ; else /tmp/update_run.sh && /etc/confluent/docker/run ; fi'"
+      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
 
   schemaregistry0:
-    image: confluentinc/cp-schema-registry:7.2.1
+    image: confluentinc/cp-schema-registry:7.6.0
     ports:
       - 8085:8085
     depends_on:
@@ -88,7 +87,7 @@ services:
     build:
       context: ./kafka-connect
       args:
-        image: confluentinc/cp-kafka-connect:6.0.1
+        image: confluentinc/cp-kafka-connect:7.6.0
     ports:
       - 8083:8083
     depends_on:
@@ -122,7 +121,7 @@ services:
   #      AWS_SECRET_ACCESS_KEY: ""
 
   kafka-init-topics:
-    image: confluentinc/cp-kafka:7.2.1
+    image: confluentinc/cp-kafka:7.6.0
     volumes:
       - ./data/message.json:/data/message.json
     depends_on:
@@ -162,7 +161,7 @@ services:
     command: bash -c '/connectors/start.sh'
 
   ksqldb:
-    image: confluentinc/ksqldb-server:0.18.0
+    image: confluentinc/cp-ksqldb-server:7.6.0
     healthcheck:
       test: [ "CMD", "timeout", "1", "curl", "--silent", "--fail", "http://localhost:8088/info" ]
       interval: 30s