more feedback

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/CommandTopicBackupImpl.java

@@ -19,7 +19,7 @@
 import com.google.common.base.Ticker;
 import io.confluent.ksql.rest.entity.CommandId;
 import io.confluent.ksql.rest.server.computation.Command;
-import io.confluent.ksql.rest.util.CommandTopicUtil;
+import io.confluent.ksql.rest.server.computation.InternalTopicSerdes;
 import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.KsqlServerException;
 import io.confluent.ksql.util.Pair;
@@ -133,8 +133,8 @@ public void writeRecord(final ConsumerRecord<byte[], byte[]> record) {
           record.topic(),
           record.partition(),
           record.offset(),
-          CommandTopicUtil.deserializeCommandId(record.topic(), record.key()),
-          CommandTopicUtil.deserializeCommand(record.topic(), record.value())
+          InternalTopicSerdes.deserializer(CommandId.class).deserialize(record.topic(), record.key()),
+          InternalTopicSerdes.deserializer(Command.class).deserialize(record.topic(), record.value())
       );
     } catch (Exception e) {
       LOG.error("Failed to deserialize command topic record when backing it up: {}:{}",

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandStore.java

@@ -49,6 +49,7 @@
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.IsolationLevel;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -307,10 +308,10 @@ private long getCommandTopicOffset() {
         COMMAND_TOPIC_PARTITION
     );
 
-    try (Consumer<CommandId, Command> commandConsumer = new KafkaConsumer<>(
+    try (Consumer<byte[], byte[]> commandConsumer = new KafkaConsumer<>(
         kafkaConsumerProperties,
-        InternalTopicSerdes.deserializer(CommandId.class),
-        InternalTopicSerdes.deserializer(Command.class)
+        new ByteArrayDeserializer(),
+        new ByteArrayDeserializer()
     )) {
       commandConsumer.assign(Collections.singleton(commandTopicPartition));
       return commandConsumer.endOffsets(Collections.singletonList(commandTopicPartition))