Tap 12766 doris empty table properties

connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/KafkaConfig.java

@@ -153,6 +153,9 @@ public String getConnectionSchemaRegisterUrl() {
         return connectionConfigGet("schemaRegisterUrl", "");
     }
 
+    public String getConnectionSchemaRegistryType() {
+        return connectionConfigGet("schemaRegistryType", "CONFLUENT");
+    }
     public String getConnectionRegistrySchemaType() {
         return connectionConfigGet("registrySchemaType", "JSON");
     }

connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/hortonworks/HortonworksSchemaRegistryClient.java

@@ -0,0 +1,175 @@
+package io.tapdata.kafka.hortonworks;
+
+import io.confluent.kafka.schemaregistry.ParsedSchema;
+import io.confluent.kafka.schemaregistry.client.SchemaMetadata;
+import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
+import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
+import okhttp3.*;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+
+/**
+ * Hortonworks Schema Registry 适配器（最小实现）
+ * 仅实现 TapData 实际用到的方法，其他方法抛 UnsupportedOperationException
+ */
+public class HortonworksSchemaRegistryClient implements SchemaRegistryClient {
+
+    private final List<String> baseUrls;
+    private final OkHttpClient httpClient;
+    private final String basicAuthHeader;
+
+    public HortonworksSchemaRegistryClient(List<String> baseUrls, Map<String, Object> configs) {
+        this.baseUrls = baseUrls;
+        this.httpClient = new OkHttpClient.Builder().build();
+
+        // 处理 Basic Auth
+        if (configs != null && "USER_INFO".equals(configs.get("basic.auth.credentials.source"))) {
+            String userInfo = (String) configs.get("basic.auth.user.info");
+            if (StringUtils.isNotBlank(userInfo)) {
+                String encoded = Base64.getEncoder().encodeToString(userInfo.getBytes(StandardCharsets.UTF_8));
+                this.basicAuthHeader = "Basic " + encoded;
+            } else {
+                this.basicAuthHeader = null;
+            }
+        } else {
+            this.basicAuthHeader = null;
+        }
+    }
+
+    @Override
+    public int register(String subject, ParsedSchema schema) throws IOException, RestClientException {
+        // Hortonworks API: POST /subjects/{subject}/versions
+        // body: {"schema": "<escaped JSON string>"}  <-- 不带 schemaType
+        String url = baseUrls.get(0) + "/subjects/" + subject + "/versions";
+        String schemaStr = schema.canonicalString();
+        // 需要转义双引号并包装成 JSON 字符串
+        String escapedSchema = schemaStr.replace("\\", "\\\\").replace("\"", "\\\"");
+        String requestBody = "{\"schema\":\"" + escapedSchema + "\"}";
+
+        Request.Builder builder = new Request.Builder()
+                .url(url)
+                .post(RequestBody.create(requestBody, MediaType.parse("application/json")));
+
+        if (basicAuthHeader != null) {
+            builder.header("Authorization", basicAuthHeader);
+        }
+
+        try (Response response = httpClient.newCall(builder.build()).execute()) {
+            String body = response.body() != null ? response.body().string() : "";
+            if (!response.isSuccessful()) {
+                throw new RestClientException("Register schema failed: " + response.code() + " " + body, 
+                        response.code(), 50001);
+            }
+            // 解析返回的 {"id": 123}
+            return parseId(body);
+        }
+    }
+
+    private int parseId(String json) {
+        // 简单解析 {"id":123}
+        int idIndex = json.indexOf("\"id\"");
+        if (idIndex < 0) {
+            throw new RuntimeException("Invalid response: " + json);
+        }
+        int colonIndex = json.indexOf(":", idIndex);
+        int commaIndex = json.indexOf(",", colonIndex);
+        int braceIndex = json.indexOf("}", colonIndex);
+        int endIndex = commaIndex > 0 ? Math.min(commaIndex, braceIndex) : braceIndex;
+        String idStr = json.substring(colonIndex + 1, endIndex).trim();
+        return Integer.parseInt(idStr);
+    }
+
+    @Override
+    public SchemaMetadata getLatestSchemaMetadata(String subject) throws IOException, RestClientException {
+        // Hortonworks API: GET /subjects/{subject}/versions/latest
+        String url = baseUrls.get(0) + "/subjects/" + subject + "/versions/latest";
+        Request.Builder builder = new Request.Builder().url(url).get();
+        if (basicAuthHeader != null) {
+            builder.header("Authorization", basicAuthHeader);
+        }
+
+        try (Response response = httpClient.newCall(builder.build()).execute()) {
+            if (response.code() == 404) {
+                throw new RestClientException("Schema not found", 404, 40401);
+            }
+            String body = response.body() != null ? response.body().string() : "";
+            if (!response.isSuccessful()) {
+                throw new RestClientException("Get schema failed: " + response.code() + " " + body,
+                        response.code(), 50001);
+            }
+            // 简化解析: 假设返回 {"id":1, "version":1, "schema":"..."}
+            int id = parseFieldInt(body, "id");
+            int version = parseFieldInt(body, "version");
+            String schema = parseFieldString(body, "schema");
+            return new SchemaMetadata(id, version, schema);
+        }
+    }
+
+    private int parseFieldInt(String json, String field) {
+        int fieldIndex = json.indexOf("\"" + field + "\"");
+        if (fieldIndex < 0) return -1;
+        int colonIndex = json.indexOf(":", fieldIndex);
+        int commaIndex = json.indexOf(",", colonIndex);
+        int braceIndex = json.indexOf("}", colonIndex);
+        int endIndex = commaIndex > 0 ? Math.min(commaIndex, braceIndex) : braceIndex;
+        String valueStr = json.substring(colonIndex + 1, endIndex).trim();
+        return Integer.parseInt(valueStr);
+    }
+
+    private String parseFieldString(String json, String field) {
+        int fieldIndex = json.indexOf("\"" + field + "\"");
+        if (fieldIndex < 0) return null;
+        int colonIndex = json.indexOf(":", fieldIndex);
+        int quoteStart = json.indexOf("\"", colonIndex);
+        int quoteEnd = json.indexOf("\"", quoteStart + 1);
+        return json.substring(quoteStart + 1, quoteEnd);
+    }
+
+    // ========== 以下是接口要求但 TapData 未使用的方法，全部抛异常 ==========
+
+    @Override
+    public List<Integer> getAllVersions(String subject) throws IOException, RestClientException {
+        throw new UnsupportedOperationException("Not implemented for Hortonworks");
+    }
+
+    @Override
+    public int getVersion(String subject, ParsedSchema schema) throws IOException, RestClientException {
+        throw new UnsupportedOperationException("Not implemented for Hortonworks");
+    }
+
+    @Override
+    public String updateCompatibility(String subject, String compatibility) throws IOException, RestClientException {
+        // 如果 TapData 用到了这个方法，需要实现
+        throw new UnsupportedOperationException("Not implemented for Hortonworks");
+    }
+
+    @Override
+    public String getCompatibility(String subject) throws IOException, RestClientException {
+        throw new UnsupportedOperationException("Not implemented for Hortonworks");
+    }
+
+    @Override
+    public Collection<String> getAllSubjects() throws IOException, RestClientException {
+        throw new UnsupportedOperationException("Not implemented for Hortonworks");
+    }
+
+    @Override
+    public int getId(String subject, ParsedSchema schema) throws IOException, RestClientException {
+        throw new UnsupportedOperationException("Not implemented for Hortonworks");
+    }
+
+    @Override
+    public ParsedSchema getSchemaById(int id) throws IOException, RestClientException {
+        throw new UnsupportedOperationException("Not implemented for Hortonworks");
+    }
+
+    @Override
+    public ParsedSchema getSchemaBySubjectAndId(String subject, int id) throws IOException, RestClientException {
+        throw new UnsupportedOperationException("Not implemented for Hortonworks");
+    }
+
+    // 还有更多接口方法，这里省略...
+}

connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/service/KafkaService.java

@@ -780,6 +780,12 @@ public SchemaRegistryClient getSchemaRegistryClient() {
                 configs.put("basic.auth.user.info",
                         config.getConnectionAuthUserName() + ":" + config.getConnectionAuthPassword());
             }
+            String registryType = config.getConnectionSchemaRegistryType();
+            if ("HORTONWORKS".equalsIgnoreCase(registryType)) {
+                schemaRegistryClient = new io.tapdata.kafka.hortonworks.HortonworksSchemaRegistryClient(urls, configs);
+                logger.info("Using Hortonworks Schema Registry adapter for URLs: " + urls);
+                return schemaRegistryClient;
+            }
             schemaRegistryClient = new CachedSchemaRegistryClient(urls, 1000, configs);
             return schemaRegistryClient;
         }

connectors/doris-connector/src/main/java/io/tapdata/connector/doris/DorisConnector.java

@@ -41,6 +41,7 @@
 import java.time.ZoneId;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -220,6 +221,23 @@ protected CreateTableOptions createDorisTable(TapConnectorContext connectorConte
             createTableOptions.setTableExists(true);
             return createTableOptions;
         }
+        createTableOptions.setTableExists(false);
+        String createTableSql = buildCreateTableSql(tapTable);
+        try {
+            tapLogger.info("Create table sql: [{}]", createTableSql);
+            dorisJdbcContext.execute(createTableSql);
+            return createTableOptions;
+        } catch (Exception e) {
+            exceptionCollector.collectWritePrivileges("createTable", Collections.emptyList(), e);
+            throw new RuntimeException("Create Table " + tapTable.getId() + " Failed | Error: " + e.getMessage() + " | Sql: " + createTableSql, e);
+        }
+    }
+
+    /**
+     * Build the CREATE TABLE sql for Doris, extracted for unit testability.
+     * Only depends on {@link DorisConfig} and {@link DorisSqlMaker}, no JDBC involved.
+     */
+    String buildCreateTableSql(TapTable tapTable) {
         Collection<String> primaryKeys = tapTable.primaryKeys(true);
         DorisTableType uniqueType = DorisTableType.valueOf(dorisConfig.getUniqueKeyType(tapTable.getId()));
         StringBuilder stringBuilder = new StringBuilder();
@@ -265,19 +283,15 @@ protected CreateTableOptions createDorisTable(TapConnectorContext connectorConte
             stringBuilder.append(String.join("`,`", dorisConfig.getDistributedKey(tapTable.getId())));
         }
         //generate bucket
-        stringBuilder.append("`) BUCKETS ").append(dorisConfig.getBucket(tapTable.getId())).append(" PROPERTIES(");
-        //generate properties
-        stringBuilder.append(dorisConfig.getTableProperties(tapTable.getId()).stream().map(v -> "\"" + v.get("propKey") + "\"=\"" + v.get("propValue") + "\"").collect(Collectors.joining(", ")));
-        stringBuilder.append(")");
-        createTableOptions.setTableExists(false);
-        try {
-            tapLogger.info("Create table sql: [{}]", stringBuilder.toString());
-            dorisJdbcContext.execute(stringBuilder.toString());
-            return createTableOptions;
-        } catch (Exception e) {
-            exceptionCollector.collectWritePrivileges("createTable", Collections.emptyList(), e);
-            throw new RuntimeException("Create Table " + tapTable.getId() + " Failed | Error: " + e.getMessage() + " | Sql: " + stringBuilder, e);
+        stringBuilder.append("`) BUCKETS ").append(dorisConfig.getBucket(tapTable.getId()));
+        //generate properties: omit the whole PROPERTIES() clause when table properties are empty, otherwise Doris parses "PROPERTIES()" as a syntax error
+        List<LinkedHashMap<String, String>> tableProperties = dorisConfig.getTableProperties(tapTable.getId());
+        if (EmptyKit.isNotEmpty(tableProperties)) {
+            stringBuilder.append(" PROPERTIES(");
+            stringBuilder.append(tableProperties.stream().map(v -> "\"" + v.get("propKey") + "\"=\"" + v.get("propValue") + "\"").collect(Collectors.joining(", ")));
+            stringBuilder.append(")");
         }
+        return stringBuilder.toString();
     }
 
 //    @Override

connectors/doris-connector/src/test/java/io/tapdata/connector/doris/DorisConnectorCreateTableSqlTest.java

@@ -0,0 +1,113 @@
+package io.tapdata.connector.doris;
+
+import io.tapdata.connector.doris.bean.DorisConfig;
+import io.tapdata.entity.schema.TapField;
+import io.tapdata.entity.schema.TapTable;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Unit tests for {@link DorisConnector#buildCreateTableSql(TapTable)}.
+ * DorisConfig is a plain POJO, TapTable is built with TapField directly,
+ * no mock or real Doris instance needed.
+ */
+public class DorisConnectorCreateTableSqlTest {
+
+    private DorisConnector connector;
+    private DorisConfig config;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        connector = new DorisConnector();
+        config = new DorisConfig();
+        config.setBucket(2);
+        config.setUniqueKeyType("Unique");
+        config.setDuplicateKey(new ArrayList<>());
+        config.setDistributedKey(new ArrayList<>());
+        config.setTableProperties(new ArrayList<>());
+        config.setSchema("test_db");
+        setField(connector, "dorisConfig", config);
+        setField(connector, "commonDbConfig", config);
+        setField(connector, "commonSqlMaker", new DorisSqlMaker());
+    }
+
+    @DisplayName("empty tableProperties should omit the whole PROPERTIES() clause")
+    @Test
+    void shouldOmitPropertiesClauseWhenTablePropertiesIsEmpty() {
+        config.setTableProperties(new ArrayList<>());
+
+        String sql = connector.buildCreateTableSql(simpleTable("test_table"));
+
+        assertTrue(sql.endsWith("BUCKETS 2"), "sql should end with BUCKETS clause, but was: " + sql);
+        assertFalse(sql.contains("PROPERTIES("), "sql should not contain PROPERTIES(), but was: " + sql);
+    }
+
+    @DisplayName("non-empty tableProperties should keep the PROPERTIES() clause as before")
+    @Test
+    void shouldKeepPropertiesClauseWhenTablePropertiesIsNotEmpty() {
+        config.setTableProperties(properties(
+                property("replication_num", "3")));
+
+        String sql = connector.buildCreateTableSql(simpleTable("test_table"));
+
+        assertTrue(sql.contains("PROPERTIES(\"replication_num\"=\"3\")"), "sql should contain PROPERTIES clause, but was: " + sql);
+    }
+
+    @DisplayName("multiple tableProperties should be joined with comma as before")
+    @Test
+    void shouldJoinMultipleTablePropertiesWithComma() {
+        config.setTableProperties(properties(
+                property("replication_num", "3"),
+                property("storage_medium", "SSD")));
+
+        String sql = connector.buildCreateTableSql(simpleTable("test_table"));
+
+        assertTrue(sql.contains("PROPERTIES(\"replication_num\"=\"3\", \"storage_medium\"=\"SSD\")"), "sql should contain joined PROPERTIES clause, but was: " + sql);
+    }
+
+    private TapTable simpleTable(String tableName) {
+        TapTable tapTable = new TapTable(tableName);
+        TapField field = new TapField("id", "VARCHAR");
+        field.setDataType("VARCHAR");
+        field.setPrimaryKeyPos(1);
+        tapTable.add(field);
+        return tapTable;
+    }
+
+    private LinkedHashMap<String, String> property(String key, String value) {
+        LinkedHashMap<String, String> map = new LinkedHashMap<>();
+        map.put("propKey", key);
+        map.put("propValue", value);
+        return map;
+    }
+
+    private List<LinkedHashMap<String, String>> properties(LinkedHashMap<String, String>... entries) {
+        List<LinkedHashMap<String, String>> list = new ArrayList<>();
+        java.util.Collections.addAll(list, entries);
+        return list;
+    }
+
+    private void setField(Object target, String fieldName, Object value) throws Exception {
+        Class<?> clazz = target.getClass();
+        while (clazz != null) {
+            try {
+                Field field = clazz.getDeclaredField(fieldName);
+                field.setAccessible(true);
+                field.set(target, value);
+                return;
+            } catch (NoSuchFieldException e) {
+                clazz = clazz.getSuperclass();
+            }
+        }
+        throw new NoSuchFieldException("field not found: " + fieldName);
+    }
+}