@Override
  public Deserializer deserializer(String topic, Target type) {
    return switch (type) {
      case KEY -> keyDeserializer();
      case VALUE -> valueDeserializer();
    };
  }
@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(ConsumerOffsetsSerde.TOPIC));
      List<Tuple2<DeserializeResult, DeserializeResult>> polled = new ArrayList<>();

      Awaitility.await()
          .pollInSameThread()
          .atMost(Duration.ofMinutes(1))
          .untilAsserted(() -> {
            for (var rec : consumer.poll(Duration.ofMillis(200))) {
              DeserializeResult key = rec.key() != null
                  ? keyDeserializer.deserialize(null, rec.key().get())
                  : null;
              DeserializeResult val = rec.value() != null
                  ? valueDeserializer.deserialize(null, rec.value().get())
                  : null;
              if (key != null && val != null) {
                polled.add(Tuples.of(key, val));
              }
            }
            assertThat(polled).anyMatch(t -> isCommitMessage(t.getT1(), t.getT2()));
            assertThat(polled).anyMatch(t -> isGroupMetadataMessage(t.getT1(), t.getT2()));
          });
    }
  }