Add ability rewrite predefined codec.

Add tests for rewrite predefined codec.
Remove ProtoUtils class

Author: ekuvardin

topic/src/main/java/tech/ydb/topic/description/Codec.java

@@ -43,17 +43,6 @@ public interface Codec {
     int GZIP = 2;
     int LZOP = 3;
     int ZSTD = 4;
-    int CUSTOM = 10000;
-
-    /**
-     * Check is codec is reserved
-     *
-     * @param codec codec id
-     * @return true - codec id is reserved; false - elsewhere
-     */
-    default boolean isReserved(int codec) {
-        return codec <= CUSTOM;
-    }
 
     /**
      * Get codec identifier

topic/src/main/java/tech/ydb/topic/description/CodecRegistry.java

@@ -39,10 +39,6 @@ public Codec registerCodec(Codec codec) {
         assert codec != null;
         int codecId = codec.getId();
 
-        if (codec.isReserved(codecId)) {
-            throw new RuntimeException(
-                    "Create custom codec for reserved code not allowed: " + codec + " .Use code more than 10000");
-        }
         Codec result = customCodecMap.put(codecId, codec);
 
         if (result != null) {

topic/src/main/java/tech/ydb/topic/description/Consumer.java

@@ -5,7 +5,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -14,7 +13,6 @@
 
 import tech.ydb.core.utils.ProtobufUtils;
 import tech.ydb.proto.topic.YdbTopic;
-import tech.ydb.topic.utils.ProtoUtils;
 
 /**
  * @author Nikolay Perfilov
@@ -40,8 +38,7 @@ public Consumer(YdbTopic.Consumer consumer) {
         this.name = consumer.getName();
         this.important = consumer.getImportant();
         this.readFrom = ProtobufUtils.protoToInstant(consumer.getReadFrom());
-        this.supportedCodecs = consumer.getSupportedCodecs().getCodecsList()
-                .stream().map(ProtoUtils::codecFromProto).collect(Collectors.toList());
+        this.supportedCodecs = new ArrayList<>(consumer.getSupportedCodecs().getCodecsList());
         this.attributes = consumer.getAttributesMap();
         this.stats = new ConsumerStats(consumer.getConsumerStats());
     }

topic/src/main/java/tech/ydb/topic/impl/TopicClientImpl.java

@@ -48,7 +48,6 @@
 import tech.ydb.topic.settings.ReadEventHandlersSettings;
 import tech.ydb.topic.settings.ReaderSettings;
 import tech.ydb.topic.settings.WriterSettings;
-import tech.ydb.topic.utils.ProtoUtils;
 import tech.ydb.topic.write.AsyncWriter;
 import tech.ydb.topic.write.SyncWriter;
 import tech.ydb.topic.write.impl.AsyncWriterImpl;
@@ -296,7 +295,7 @@ private TopicDescription mapDescribeTopic(YdbTopic.DescribeTopicResult result) {
 
         SupportedCodecs.Builder supportedCodecsBuilder = SupportedCodecs.newBuilder();
         for (int codec : result.getSupportedCodecs().getCodecsList()) {
-            supportedCodecsBuilder.addCodec(ProtoUtils.codecFromProto(codec));
+            supportedCodecsBuilder.addCodec(codec);
         }
         description.setSupportedCodecs(supportedCodecsBuilder.build());
 
@@ -383,7 +382,7 @@ private static YdbTopic.Consumer toProto(Consumer consumer) {
         List<Integer> supportedCodecs = consumer.getSupportedCodecsList();
         if (!supportedCodecs.isEmpty()) {
             YdbTopic.SupportedCodecs.Builder codecBuilder = YdbTopic.SupportedCodecs.newBuilder();
-            supportedCodecs.forEach(codec -> codecBuilder.addCodecs(ProtoUtils.toProto(codec)));
+            supportedCodecs.forEach(codecBuilder::addCodecs);
             consumerBuilder.setSupportedCodecs(codecBuilder.build());
         }
 
@@ -394,7 +393,7 @@ private static YdbTopic.SupportedCodecs toProto(SupportedCodecs supportedCodecs)
         List<Integer> supportedCodecsList = supportedCodecs.getCodecs();
         YdbTopic.SupportedCodecs.Builder codecsBuilder = YdbTopic.SupportedCodecs.newBuilder();
         for (Integer codec : supportedCodecsList) {
-            codecsBuilder.addCodecs(tech.ydb.topic.utils.ProtoUtils.toProto(codec));
+            codecsBuilder.addCodecs(codec);
         }
         return codecsBuilder.build();
     }

topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java

@@ -18,7 +18,7 @@ public class BatchMeta {
     public BatchMeta(YdbTopic.StreamReadMessage.ReadResponse.Batch batch) {
         this.producerId = batch.getProducerId();
         this.writeSessionMeta = batch.getWriteSessionMetaMap();
-        this.codec = tech.ydb.topic.utils.ProtoUtils.codecFromProto(batch.getCodec());
+        this.codec = batch.getCodec();
         this.writtenAt = ProtobufUtils.protoToInstant(batch.getWrittenAt());
     }
 

topic/src/main/java/tech/ydb/topic/utils/ProtoUtils.java

@@ -15,7 +15,6 @@
 import tech.ydb.proto.topic.YdbTopic;
 import tech.ydb.topic.description.MetadataItem;
 import tech.ydb.topic.settings.WriterSettings;
-import tech.ydb.topic.utils.ProtoUtils;
 
 /**
  * Utility class that splits messages into several requests so that every request would be less than grpc size limit
@@ -80,7 +79,7 @@ public void setSession(WriteSession session) {
 
     private void reset() {
         writeRequestBuilder = YdbTopic.StreamWriteMessage.WriteRequest.newBuilder()
-                .setCodec(ProtoUtils.toProto(settings.getCodec()));
+                .setCodec(settings.getCodec());
         messageCount = 0;
         totalMessageDataProtoSize = 0;
     }
@@ -124,7 +123,7 @@ public void sendWriteRequest() {
                         messages.subList(firstHalfMessagesCount, messages.size())
                 )) {
                     writeRequestBuilder = YdbTopic.StreamWriteMessage.WriteRequest.newBuilder()
-                                .setCodec(ProtoUtils.toProto(settings.getCodec()));
+                                .setCodec(settings.getCodec());
                     writeRequestBuilder.addAllMessages(sublist);
                     YdbTopic.StreamWriteMessage.FromClient subRequest = YdbTopic.StreamWriteMessage.FromClient
                             .newBuilder()

topic/src/main/java/tech/ydb/topic/write/impl/MessageSender.java

@@ -16,7 +16,7 @@
  *
  * @author Evgeny Kuvardin
  */
-public class YdbTopicsCustomCodecImplTest {
+public class CodecRegistryTest {
     CodecRegistry registry;
 
     private static final int codecId = 10113;
@@ -46,25 +46,18 @@ public void registerCustomCodecShouldNotAcceptNull() {
     }
 
     @Test
-    public void registerCustomCodecShouldFailedWhenRegisterReservedCode() {
+    public void registerCustomCodecShouldRegisterAndOverrideAnyCodec() {
         CodecTopic codec1 = new CodecTopic();
-        expectErrorRegister(-1, codec1);
-        expectErrorRegister(-100, codec1);
-        expectErrorRegister(0, codec1);
-        expectErrorRegister(1, codec1);
-        expectErrorRegister(2, codec1);
-        expectErrorRegister(3, codec1);
-        expectErrorRegister(4, codec1);
-        expectErrorRegister(10000, codec1);
+        expectRegisterCodec(1, codec1, RawCodec.getInstance());
+        expectRegisterCodec(2, codec1, GzipCodec.getInstance());
+        expectRegisterCodec(3, codec1, LzopCodec.getInstance());
+        expectRegisterCodec(4, codec1, ZstdCodec.getInstance());
     }
 
-    void expectErrorRegister(int codecId, CodecTopic codec) {
-        codec.setCodecId(codecId);
-        Exception e = Assert.assertThrows(
-                RuntimeException.class,
-                () -> registry.registerCodec(codec));
-
-        Assert.assertEquals("Create custom codec for reserved code not allowed: " + codec + " .Use code more than 10000", e.getMessage());
+    void expectRegisterCodec(int codecId, CodecTopic newCodec, Codec oldCodec) {
+        newCodec.setCodecId(codecId);
+        Codec codecOldPredefined = registry.registerCodec(newCodec);
+        Assert.assertSame(codecOldPredefined, oldCodec);
     }
 
     static class CodecTopic implements Codec {

topic/src/test/java/tech/ydb/topic/impl/YdbTopicsCustomCodecImplTest.java renamed to topic/src/test/java/tech/ydb/topic/impl/CodecRegistryTest.java

@@ -305,10 +305,9 @@ public void writeWithReservedNotExistedCodec() {
         createTopic(client1, TEST_TOPIC1);
 
         Exception e = Assert.assertThrows(RuntimeException.class, () -> writeData(7, TEST_TOPIC1, client1));
-        Assert.assertEquals("Cannot convert codec to proto. Unknown codec value: " + 7, e.getMessage());
+        Assert.assertTrue(e.getMessage().contains("Unsupported codec: " + 7));
     }
 
-
     /**
      * Create one more defect. Test failed for unknown reason. Seems RuntimeException produce some weird behaviour
      */
@@ -340,6 +339,33 @@ public void readWriteRawCodec() throws ExecutionException, InterruptedException,
         readData(TEST_TOPIC1, client1);
     }
 
+    /**
+     * The test checks that we can rewrite the predefined RAW codec.
+     * Please note that modifying a RAW codec is highly unusual and potentially risky.
+     * You take full responsibility for any consequences that may result.
+     * The SDK includes mechanisms in some parts of the codec that attempt to optimize the code
+     * and detect write or read operations to RAW codecs.
+     * <p>
+     * 1. Create client1
+     * 2. Create topic TEST_TOPIC1 in client1
+     * 3. Create custom codec
+     * 4, Register codec
+     * 5. Try to write
+     * 6. Read data
+     */
+    @Test
+    public void userCanRewriteRawCodec() throws ExecutionException, InterruptedException, TimeoutException {
+        client1 = createClient();
+        createTopic(client1, TEST_TOPIC1);
+
+        Codec codec = new CustomCodec(0, Codec.RAW);
+        client1.registerCodec(codec);
+
+        writeData(codec.getId(), TEST_TOPIC1, client1);
+
+        readData(TEST_TOPIC1, client1);
+    }
+
     /**
      * Test checks that we can write and read using GZIP Codec
      * <p>
@@ -358,6 +384,29 @@ public void readWriteGzipCodec() throws ExecutionException, InterruptedException
         readData(TEST_TOPIC1, client1);
     }
 
+    /**
+     * The test checks that we can rewrite the predefined Gzip codec.
+     * <p>
+     * 1. Create client1
+     * 2. Create topic TEST_TOPIC1 in client1
+     * 3. Create custom codec
+     * 4, Register codec
+     * 5. Try to write
+     * 6. Read data
+     */
+    @Test
+    public void userCanRewriteGzipCodec() throws ExecutionException, InterruptedException, TimeoutException {
+        client1 = createClient();
+        createTopic(client1, TEST_TOPIC1);
+
+        Codec codec = new CustomCodec(2, Codec.GZIP);
+        client1.registerCodec(codec);
+
+        writeData(codec.getId(), TEST_TOPIC1, client1);
+
+        readData(TEST_TOPIC1, client1);
+    }
+
     /**
      * Test checks that we can write and read using Lzop Codec
      * <p>
@@ -376,6 +425,29 @@ public void readWriteLzopCodec() throws ExecutionException, InterruptedException
         readData(TEST_TOPIC1, client1);
     }
 
+    /**
+     * The test checks that we can rewrite the predefined Lzop codec.
+     * <p>
+     * 1. Create client1
+     * 2. Create topic TEST_TOPIC1 in client1
+     * 3. Create custom codec
+     * 4, Register codec
+     * 5. Try to write
+     * 6. Read data
+     */
+    @Test
+    public void userCanRewriteLzopCodec() throws ExecutionException, InterruptedException, TimeoutException {
+        client1 = createClient();
+        createTopic(client1, TEST_TOPIC1);
+
+        Codec codec = new CustomCodec(3, Codec.LZOP);
+        client1.registerCodec(codec);
+
+        writeData(codec.getId(), TEST_TOPIC1, client1);
+
+        readData(TEST_TOPIC1, client1);
+    }
+
     /**
      * Test checks that we can write and read using Zstd Codec
      * <p>
@@ -394,6 +466,29 @@ public void readWriteZstdCodec() throws ExecutionException, InterruptedException
         readData(TEST_TOPIC1, client1);
     }
 
+    /**
+     * The test checks that we can rewrite the predefined Lzop codec.
+     * <p>
+     * 1. Create client1
+     * 2. Create topic TEST_TOPIC1 in client1
+     * 3. Create custom codec
+     * 4, Register codec
+     * 5. Try to write
+     * 6. Read data
+     */
+    @Test
+    public void userCanRewriteZstdCodec() throws ExecutionException, InterruptedException, TimeoutException {
+        client1 = createClient();
+        createTopic(client1, TEST_TOPIC1);
+
+        Codec codec = new CustomCodec(4, Codec.ZSTD);
+        client1.registerCodec(codec);
+
+        writeData(codec.getId(), TEST_TOPIC1, client1);
+
+        readData(TEST_TOPIC1, client1);
+    }
+
     private TopicClient createClient() {
         TopicClient topicClient = TopicClient.newClient(ydbTransport).build();
         clientToClose.add(topicClient);
@@ -468,7 +563,7 @@ private void readData(String topicName, TopicClient client) throws InterruptedEx
 
             Assert.assertNotNull(testMessages);
             for (byte[] bytes : testMessages) {
-                tech.ydb.topic.read.Message msg = reader.receive(1, TimeUnit.SECONDS);
+                tech.ydb.topic.read.Message msg = reader.receive(10, TimeUnit.SECONDS);
                 Assert.assertNotNull(msg);
                 Assert.assertArrayEquals(bytes, msg.getData());
             }