diff --git a/connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/KafkaConfig.java b/connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/KafkaConfig.java index 98f1322db..4edc424a5 100644 --- a/connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/KafkaConfig.java +++ b/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"); } diff --git a/connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/hortonworks/HortonworksSchemaRegistryClient.java b/connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/hortonworks/HortonworksSchemaRegistryClient.java new file mode 100644 index 000000000..99c4d3635 --- /dev/null +++ b/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 baseUrls; + private final OkHttpClient httpClient; + private final String basicAuthHeader; + + public HortonworksSchemaRegistryClient(List baseUrls, Map 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": ""} <-- 不带 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 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 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"); + } + + // 还有更多接口方法，这里省略... +} diff --git a/connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/service/KafkaService.java b/connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/service/KafkaService.java index e3db94573..a41d95f88 100644 --- a/connectors-common/kafka-enhanced-core/src/main/java/io/tapdata/kafka/service/KafkaService.java +++ b/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; } diff --git a/connectors/doris-connector/src/main/java/io/tapdata/connector/doris/DorisConnector.java b/connectors/doris-connector/src/main/java/io/tapdata/connector/doris/DorisConnector.java index c07ce3b63..ece33a287 100644 --- a/connectors/doris-connector/src/main/java/io/tapdata/connector/doris/DorisConnector.java +++ b/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 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> 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 diff --git a/connectors/doris-connector/src/test/java/io/tapdata/connector/doris/DorisConnectorCreateTableSqlTest.java b/connectors/doris-connector/src/test/java/io/tapdata/connector/doris/DorisConnectorCreateTableSqlTest.java new file mode 100644 index 000000000..21aef9cba --- /dev/null +++ b/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 property(String key, String value) { + LinkedHashMap map = new LinkedHashMap<>(); + map.put("propKey", key); + map.put("propValue", value); + return map; + } + + private List> properties(LinkedHashMap... entries) { + List> 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); + } +}