tests added

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/builtin/ConsumerOffsetsSerde.java

@@ -149,7 +149,7 @@ public class ConsumerOffsetsSerde implements BuiltInSerde {
             new Field("commit_timestamp", Type.INT64, "")
         );
 
-    final Schema commitOffsetSchemaV4Latest = new Schema(
+    final Schema commitOffsetSchemaV4 = new Schema(
         new Field("offset", Type.INT64, ""),
         new Field("leader_epoch", Type.INT32, ""),
         new Field("metadata", Type.COMPACT_STRING, ""),
@@ -227,7 +227,7 @@ public class ConsumerOffsetsSerde implements BuiltInSerde {
             )), "")
         );
 
-    final Schema metadataSchema4Latest =
+    final Schema metadataSchema4 =
         new Schema(
             new Field("protocol_type", Type.COMPACT_STRING, ""),
             new Field("generation", Type.INT32, ""),
@@ -259,7 +259,8 @@ public class ConsumerOffsetsSerde implements BuiltInSerde {
               case 1 -> metadataSchema1.read(bb);
               case 2 -> metadataSchema2.read(bb);
               case 3 -> metadataSchema3.read(bb);
-              default -> metadataSchema4Latest.read(bb);
+              case 4 -> metadataSchema4.read(bb);
+              default -> throw new IllegalArgumentException("Unrecognized version: " + version);
             }
         );
       } catch (Throwable e) {
@@ -271,7 +272,8 @@ public class ConsumerOffsetsSerde implements BuiltInSerde {
               case 1 -> commitOffsetSchemaV1.read(bb);
               case 2 -> commitOffsetSchemaV2.read(bb);
               case 3 -> commitOffsetSchemaV3.read(bb);
-              default -> commitOffsetSchemaV4Latest.read(bb);
+              case 4 -> commitOffsetSchemaV4.read(bb);
+              default -> throw new IllegalArgumentException("Unrecognized version: " + version);
             }
         );
       }

kafka-ui-api/src/test/java/com/provectus/kafka/ui/serdes/builtin/ConsumerOffsetsSerdeTest.java

@@ -21,10 +21,9 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.serialization.BytesDeserializer;
 import org.apache.kafka.common.utils.Bytes;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
-import org.testcontainers.shaded.org.apache.commons.lang3.mutable.MutableBoolean;
 import org.testcontainers.shaded.org.awaitility.Awaitility;
 import reactor.util.function.Tuple2;
 import reactor.util.function.Tuples;
@@ -33,11 +32,11 @@ class ConsumerOffsetsSerdeTest extends AbstractIntegrationTest {
 
   private static final int MSGS_TO_GENERATE = 10;
 
-  private String consumerGroupName;
-  private String committedTopic;
+  private static String consumerGroupName;
+  private static String committedTopic;
 
-  @BeforeEach
-  void createTopicAndCommitItsOffset() {
+  @BeforeAll
+  static void createTopicAndCommitItsOffset() {
     committedTopic = ConsumerOffsetsSerdeTest.class.getName() + "-" + UUID.randomUUID();
     consumerGroupName = committedTopic + "-group";
     createTopic(new NewTopic(committedTopic, 1, (short) 1));
@@ -57,19 +56,28 @@ class ConsumerOffsetsSerdeTest extends AbstractIntegrationTest {
     }
   }
 
-  @AfterEach
-  void cleanUp() {
+  @AfterAll
+  static void cleanUp() {
     deleteTopic(committedTopic);
   }
 
   @Test
-  void test() {
+  void canOnlyDeserializeConsumerOffsetsTopic() {
+    var serde = new ConsumerOffsetsSerde();
+    assertThat(serde.canDeserialize(ConsumerOffsetsSerde.TOPIC, Serde.Target.KEY)).isTrue();
+    assertThat(serde.canDeserialize(ConsumerOffsetsSerde.TOPIC, Serde.Target.VALUE)).isTrue();
+    assertThat(serde.canDeserialize("anyOtherTopic", Serde.Target.KEY)).isFalse();
+    assertThat(serde.canDeserialize("anyOtherTopic", Serde.Target.VALUE)).isFalse();
+  }
+
+  @Test
+  void deserializesMessagesMadeByConsumerActivity() {
     var serde = new ConsumerOffsetsSerde();
     var keyDeserializer = serde.deserializer(TOPIC, Serde.Target.KEY);
     var valueDeserializer = serde.deserializer(TOPIC, Serde.Target.VALUE);
 
     try (var consumer = createConsumer(consumerGroupName + "-check")) {
-      consumer.subscribe(List.of(TOPIC));
+      consumer.subscribe(List.of(ConsumerOffsetsSerde.TOPIC));
       List<Tuple2<DeserializeResult, DeserializeResult>> polled = new ArrayList<>();
 
       Awaitility.await()
@@ -158,7 +166,7 @@ class ConsumerOffsetsSerdeTest extends AbstractIntegrationTest {
     return new JsonMapper().readValue(result.getResult(), Map.class);
   }
 
-  private KafkaConsumer<Bytes, Bytes> createConsumer(String groupId) {
+  private static KafkaConsumer<Bytes, Bytes> createConsumer(String groupId) {
     Properties props = new Properties();
     props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
     props.put(ConsumerConfig.CLIENT_ID_CONFIG, groupId);