Reduced duplicate code

Author: germanosin

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

@@ -169,48 +169,21 @@ private void registerConsumerOffsetsSerde(Map<String, SerdeInstance> serdes) {
   }
 
   private void registerMirrorMakerSerdes(Map<String, SerdeInstance> serdes) {
-    registerHeartbeatSerde(serdes);
-    registerOffsetSyncSerde(serdes);
-    registerCheckpointSerde(serdes);
-  }
-
-  private void registerHeartbeatSerde(Map<String, SerdeInstance> serdes) {
-    serdes.put(
-        HeartbeatSerde.name(),
-        new SerdeInstance(
-            HeartbeatSerde.name(),
-            new HeartbeatSerde(),
-            HeartbeatSerde.TOPIC_NAME_PATTERN,
-            HeartbeatSerde.TOPIC_NAME_PATTERN,
-            null
-        )
-    );
-  }
-
-  private void registerOffsetSyncSerde(Map<String, SerdeInstance> serdes) {
-    serdes.put(
-        OffsetSyncSerde.name(),
-        new SerdeInstance(
-            OffsetSyncSerde.name(),
-            new OffsetSyncSerde(),
-            OffsetSyncSerde.TOPIC_NAME_PATTERN,
-            OffsetSyncSerde.TOPIC_NAME_PATTERN,
-            null
-        )
+    Map<String,Map.Entry<Pattern, BuiltInSerde>> mmSerdes = Map.of(
+        HeartbeatSerde.name(), Map.entry(HeartbeatSerde.TOPIC_NAME_PATTERN, new HeartbeatSerde()),
+        OffsetSyncSerde.name(), Map.entry(OffsetSyncSerde.TOPIC_NAME_PATTERN, new OffsetSyncSerde()),
+        CheckpointSerde.name(), Map.entry(CheckpointSerde.TOPIC_NAME_PATTERN, new CheckpointSerde())
     );
-  }
+    for (Map.Entry<String, Map.Entry<Pattern, BuiltInSerde>> serde : mmSerdes.entrySet()) {
+      String name = serde.getKey();
+      Pattern pattern = serde.getValue().getKey();
+      BuiltInSerde serdeInstance = serde.getValue().getValue();
 
-  private void registerCheckpointSerde(Map<String, SerdeInstance> serdes) {
-    serdes.put(
-        CheckpointSerde.name(),
-        new SerdeInstance(
-            CheckpointSerde.name(),
-            new CheckpointSerde(),
-            CheckpointSerde.TOPIC_NAME_PATTERN,
-            CheckpointSerde.TOPIC_NAME_PATTERN,
-            null
-        )
-    );
+      serdes.put(
+          name,
+          new SerdeInstance(name, serdeInstance, pattern, pattern, null)
+      );
+    }
   }
 
   private SerdeInstance createFallbackSerde() {

api/src/main/java/io/kafbat/ui/serdes/builtin/ConsumerOffsetsSerde.java

@@ -23,9 +23,7 @@
 
 // Deserialization logic and message's schemas can be found in
 // kafka.coordinator.group.GroupMetadataManager (readMessageKey, readOffsetMessageValue, readGroupMessageValue)
-public class ConsumerOffsetsSerde implements BuiltInSerde {
-
-  private static final JsonMapper JSON_MAPPER = createMapper();
+public class ConsumerOffsetsSerde extends StructSerde implements BuiltInSerde {
 
   private static final String ASSIGNMENT = "assignment";
   private static final String CLIENT_HOST = "client_host";
@@ -50,24 +48,6 @@ public static String name() {
     return "__consumer_offsets";
   }
 
-  private static JsonMapper createMapper() {
-    var module = new SimpleModule();
-    module.addSerializer(Struct.class, new JsonSerializer<>() {
-      @Override
-      public void serialize(Struct value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
-        gen.writeStartObject();
-        for (BoundField field : value.schema().fields()) {
-          var fieldVal = value.get(field);
-          gen.writeObjectField(field.def.name, fieldVal);
-        }
-        gen.writeEndObject();
-      }
-    });
-    var mapper = new JsonMapper();
-    mapper.registerModule(module);
-    return mapper;
-  }
-
   @Override
   public Optional<String> getDescription() {
     return Optional.empty();
@@ -304,8 +284,5 @@ private Deserializer valueDeserializer() {
     };
   }
 
-  @SneakyThrows
-  private String toJson(Struct s) {
-    return JSON_MAPPER.writeValueAsString(s);
-  }
+
 }

api/src/main/java/io/kafbat/ui/serdes/builtin/StructSerde.java

@@ -0,0 +1,39 @@
+package io.kafbat.ui.serdes.builtin;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import java.io.IOException;
+import lombok.SneakyThrows;
+import org.apache.kafka.common.protocol.types.BoundField;
+import org.apache.kafka.common.protocol.types.Struct;
+
+public abstract class StructSerde {
+
+  private static final JsonMapper JSON_MAPPER = createMapper();
+
+  private static JsonMapper createMapper() {
+    var module = new SimpleModule();
+    module.addSerializer(Struct.class, new JsonSerializer<>() {
+      @Override
+      public void serialize(Struct value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+        gen.writeStartObject();
+        for (BoundField field : value.schema().fields()) {
+          var fieldVal = value.get(field);
+          gen.writeObjectField(field.def.name, fieldVal);
+        }
+        gen.writeEndObject();
+      }
+    });
+    var mapper = new JsonMapper();
+    mapper.registerModule(module);
+    return mapper;
+  }
+
+  @SneakyThrows
+  protected String toJson(Struct s) {
+    return JSON_MAPPER.writeValueAsString(s);
+  }
+}

api/src/main/java/io/kafbat/ui/serdes/builtin/mm2/CheckpointSerde.java

@@ -2,7 +2,6 @@
 
 import io.kafbat.ui.serdes.BuiltInSerde;
 import java.util.Optional;
-import java.util.OptionalInt;
 import java.util.regex.Pattern;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.common.protocol.types.Field;

api/src/main/java/io/kafbat/ui/serdes/builtin/mm2/MirrorMakerSerde.java

@@ -1,46 +1,19 @@
 package io.kafbat.ui.serdes.builtin.mm2;
 
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.databind.JsonSerializer;
-import com.fasterxml.jackson.databind.SerializerProvider;
-import com.fasterxml.jackson.databind.json.JsonMapper;
-import com.fasterxml.jackson.databind.module.SimpleModule;
 import io.kafbat.ui.serde.api.DeserializeResult;
 import io.kafbat.ui.serde.api.SchemaDescription;
 import io.kafbat.ui.serde.api.Serde;
 import io.kafbat.ui.serdes.BuiltInSerde;
-import java.io.IOException;
+import io.kafbat.ui.serdes.builtin.StructSerde;
 import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.Optional;
 import lombok.RequiredArgsConstructor;
-import lombok.SneakyThrows;
-import org.apache.kafka.common.protocol.types.BoundField;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
 
 @RequiredArgsConstructor
-abstract class MirrorMakerSerde implements BuiltInSerde {
-
-  protected static final JsonMapper JSON_MAPPER = createMapper();
-
-  protected static JsonMapper createMapper() {
-    var module = new SimpleModule();
-    module.addSerializer(Struct.class, new JsonSerializer<>() {
-      @Override
-      public void serialize(Struct value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
-        gen.writeStartObject();
-        for (BoundField field : value.schema().fields()) {
-          var fieldVal = value.get(field);
-          gen.writeObjectField(field.def.name, fieldVal);
-        }
-        gen.writeEndObject();
-      }
-    });
-    var mapper = new JsonMapper();
-    mapper.registerModule(module);
-    return mapper;
-  }
+abstract class MirrorMakerSerde extends StructSerde implements BuiltInSerde {
 
   protected final boolean versioned;
 
@@ -69,11 +42,6 @@ public Serde.Serializer serializer(String topic, Serde.Target type) {
     throw new UnsupportedOperationException();
   }
 
-  @SneakyThrows
-  protected static String toJson(Struct s) {
-    return JSON_MAPPER.writeValueAsString(s);
-  }
-
   @Override
   public Deserializer deserializer(String topic, Target target) {
     return (recordHeaders, bytes) ->

api/src/main/java/io/kafbat/ui/serdes/builtin/mm2/OffsetSyncSerde.java

@@ -2,7 +2,6 @@
 
 import io.kafbat.ui.serdes.BuiltInSerde;
 import java.util.Optional;
-import java.util.OptionalInt;
 import java.util.regex.Pattern;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.common.protocol.types.Field;

api/src/test/java/io/kafbat/ui/serdes/builtin/mm2/HeartbeatSerdeTest.java

@@ -46,14 +46,15 @@ void testDeserializeKey() throws JsonProcessingException {
   }
 
   @Test
-  void testDeserializeValue() {
+  void testDeserializeValue() throws JsonProcessingException {
     var value = decodeBase64("AAAAAAGZgCEMZA==");
-    var expected = "1758791273572";
+    var expected = Map.of("timestamp", 1758791273572L);
 
     var result = SERDE.deserializer(TOPIC, Serde.Target.VALUE).deserialize(HEADERS, value);
+    var resultMap = jsonToMap(result.getResult());
 
-    assertEquals(DeserializeResult.Type.STRING, result.getType());
-    assertEquals(expected, result.getResult());
+    assertEquals(DeserializeResult.Type.JSON, result.getType());
+    assertEquals(expected, resultMap);
   }
 
 }