merge converters for change event and snapshot and use real encoding for change event

Author: whhe

docs/content/connectors/oceanbase-cdc(ZH).md

@@ -252,9 +252,10 @@ public class OceanBaseSourceExample {
                       .tableName("test_table")
                       .hostname("127.0.0.1")
                       .port(2881)
+                      .jdbcDriver("com.mysql.jdbc.Driver")
                       .logProxyHost("127.0.0.1")
                       .logProxyPort(2983)
-                      .serverTimeZone(serverTimezone)
+                      .serverTimeZone(serverTimeZone)
                       .deserializer(deserializer)
                       .build();
 

docs/content/connectors/oceanbase-cdc.md

@@ -430,9 +430,10 @@ public class OceanBaseSourceExample {
                       .tableName("test_table")
                       .hostname("127.0.0.1")
                       .port(2881)
+                      .jdbcDriver("com.mysql.jdbc.Driver")
                       .logProxyHost("127.0.0.1")
                       .logProxyPort(2983)
-                      .serverTimeZone(serverTimezone)
+                      .serverTimeZone(serverTimeZone)
                       .deserializer(deserializer)
                       .build();
 

flink-connector-oceanbase-cdc/src/main/java/com/ververica/cdc/connectors/oceanbase/source/OceanBaseDeserializationRuntimeConverter.java

@@ -16,32 +16,13 @@
 
 package com.ververica.cdc.connectors.oceanbase.source;
 
-import com.oceanbase.oms.logmessage.ByteString;
-import sun.reflect.generics.reflectiveObjects.NotImplementedException;
-
 import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
 
 /**
  * Runtime converter that converts objects of OceanBase into objects of Flink Table & SQL internal
  * data structures.
  */
 public interface OceanBaseDeserializationRuntimeConverter extends Serializable {
 
-    default Object convert(Object object) throws Exception {
-        if (object instanceof ByteString) {
-            return convertChangeEvent(
-                    ((ByteString) object).toString(StandardCharsets.UTF_8.name()));
-        } else {
-            return convertSnapshotEvent(object);
-        }
-    }
-
-    default Object convertSnapshotEvent(Object object) throws Exception {
-        throw new NotImplementedException();
-    }
-
-    default Object convertChangeEvent(String string) throws Exception {
-        throw new NotImplementedException();
-    }
+    Object convert(Object object);
 }