Merge remote-tracking branch 'origin/365.Add-support-of-custom-codecs-in-topics' into 365.Add-support-of-custom-codecs-in-topics

# Conflicts:
#	topic/src/main/java/tech/ydb/topic/utils/Encoder.java
#	topic/src/main/java/tech/ydb/topic/write/impl/AsyncWriterImpl.java
#	topic/src/main/java/tech/ydb/topic/write/impl/SyncWriterImpl.java
#	topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java
#	topic/src/test/java/tech/ydb/topic/impl/YdbTopicsCustomCodecIntegrationTest.java
#	topic/src/test/java/tech/ydb/topic/impl/YdbTopicsCustomCodecTest.java

Author: ekuvardin

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

@@ -0,0 +1,48 @@
+package tech.ydb.topic.description;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Register for custom topic codec. Local to TopicClient
+ *
+ * @author Evgeny Kuvardin
+ **/
+public class CodecRegistryImpl implements CodecRegistry {
+
+    /**
+     * Make customCodecMap concurrent since register/unregister/read can be from different threads
+     */
+    final Map<Integer, CustomTopicCodec> customCodecMap;
+
+    public CodecRegistryImpl() {
+        customCodecMap = new ConcurrentHashMap<>();
+    }
+
+    @Override
+    public CustomTopicCodec registerCustomCodec(int codec, CustomTopicCodec customTopicCodec) {
+        assert customTopicCodec != null;
+
+        if (Codec.getInstance().isReserved(codec)) {
+            throw new RuntimeException(
+                    "Create custom codec for reserved code not allowed: " + codec + " .Use code more than 10000");
+        }
+
+        return customCodecMap.put(codec, customTopicCodec);
+    }
+
+    @Override
+    public CustomTopicCodec unregisterCustomCodec(int codec) {
+        if (Codec.getInstance().isReserved(codec)) {
+            throw new RuntimeException(
+                    "Create custom codec for reserved code not allowed: " + codec + " .Use code more than 10000");
+        }
+
+        return customCodecMap.remove(codec);
+    }
+
+    @Override
+    public CustomTopicCodec getCustomCodec(int codec) {
+        return customCodecMap.get(codec);
+    }
+}

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

@@ -24,6 +24,7 @@
 import tech.ydb.topic.TopicClient;
 import tech.ydb.topic.TopicRpc;
 import tech.ydb.topic.description.CodecRegistry;
+import tech.ydb.topic.description.CodecRegistryImpl;
 import tech.ydb.topic.description.Consumer;
 import tech.ydb.topic.description.ConsumerDescription;
 import tech.ydb.topic.description.CustomTopicCodec;