yuzelin commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1299903152
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java:
##########
@@ -369,4 +372,19 @@ protected FileStoreTable getFileStoreTable(String
tableName) throws Exception {
Identifier identifier = Identifier.create(database, tableName);
return (FileStoreTable) catalog().getTable(identifier);
}
+
+ protected void waitTablesCreated(String... tables) throws Exception {
+ CommonTestUtils.waitUtil(
+ () -> {
+ try {
+ List<String> existed = catalog().listTables(database);
+ return existed.containsAll(Arrays.asList(tables));
+ } catch (Catalog.DatabaseNotExistException e) {
+ throw new RuntimeException(e);
+ }
+ },
+ Duration.ofSeconds(50),
+ Duration.ofMillis(100),
+ "Failed to wait tables to be created in 5 seconds.");
Review Comment:
Error message doesn't match the code.
##########
docs/content/how-to/cdc-ingestion.md:
##########
@@ -294,11 +294,15 @@ If a message in a Kafka topic is a change event captured
from another database u
</tr>
<tr>
<td><a
href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/ogg/";>OGG
CDC</a></td>
- <td>False</td>
+ <td>True</td>
</tr>
</tbody>
</table>
+**Note that**
+
+In Oracle GoldenGate, the data format synchronized to Kafka does not include
field data type information. As a result, Paimon sets the data type for all
fields to "String" by default.
+
Review Comment:
Use
```
{{< hint info >}}
information here
{{< /hint >}}
```
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchema.java:
##########
@@ -117,16 +120,13 @@ public static KafkaSchema getKafkaSchema(Configuration
kafkaConfig, String topic
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(POLL_TIMEOUT_MILLIS));
for (ConsumerRecord<String, String> record : records) {
- String format =
kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT);
- if ("canal-json".equals(format)) {
- CanalRecordParser parser = new CanalRecordParser(true,
Collections.emptyList());
- KafkaSchema kafkaSchema =
parser.getKafkaSchema(record.value());
- if (kafkaSchema != null) {
- return kafkaSchema;
- }
- } else {
- throw new UnsupportedOperationException(
- "This format: " + format + " is not support.");
+ DataFormat format = getDataFormat(kafkaConfig);
+ RecordParser recordParser =
+ format.createParser(
+ true, new TableNameConverter(true),
Collections.emptyList());
Review Comment:
This part can be moved before `while (true)`
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.kafka.formats.ogg;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSchema;
+import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.sink.cdc.CdcRecord;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Implementation of {@link RecordParser} for parsing messages in the Ogg
format.
+ *
+ * <p>This parser handles records in the Ogg format and extracts relevant
information to produce
+ * {@link RichCdcMultiplexRecord} objects.
+ */
+public class OggRecordParser extends RecordParser {
+
+ private static final String FIELD_PRIMARY_KEYS = "primary_keys";
+ private static final String FIELD_BEFORE = "before";
+ private static final String FIELD_AFTER = "after";
+ private static final String FIELD_TYPE = "op_type";
+ private static final String OP_UPDATE = "U";
+ private static final String OP_INSERT = "I";
+ private static final String OP_DELETE = "D";
+ private final List<ComputedColumn> computedColumns;
+
+ public OggRecordParser(
+ boolean caseSensitive,
+ TableNameConverter tableNameConverter,
+ List<ComputedColumn> computedColumns) {
+ super(tableNameConverter, caseSensitive);
+ this.computedColumns = computedColumns;
+ }
+
+ @Override
+ public List<RichCdcMultiplexRecord> extractRecords() {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
+ String type = extractString(FIELD_TYPE);
+ switch (type) {
+ case OP_UPDATE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.UPDATE_BEFORE,
records);
+ processRecord(root.get(FIELD_AFTER), RowKind.UPDATE_AFTER,
records);
Review Comment:
CanalRecordParser and MySqlDebeziumJsonEventParser set the records to -D and
+I. I think here should stay consistent.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.kafka.formats.ogg;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSchema;
+import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.sink.cdc.CdcRecord;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Implementation of {@link RecordParser} for parsing messages in the Ogg
format.
+ *
+ * <p>This parser handles records in the Ogg format and extracts relevant
information to produce
+ * {@link RichCdcMultiplexRecord} objects.
+ */
+public class OggRecordParser extends RecordParser {
+
+ private static final String FIELD_PRIMARY_KEYS = "primary_keys";
+ private static final String FIELD_BEFORE = "before";
+ private static final String FIELD_AFTER = "after";
+ private static final String FIELD_TYPE = "op_type";
+ private static final String OP_UPDATE = "U";
+ private static final String OP_INSERT = "I";
+ private static final String OP_DELETE = "D";
+ private final List<ComputedColumn> computedColumns;
+
+ public OggRecordParser(
+ boolean caseSensitive,
+ TableNameConverter tableNameConverter,
+ List<ComputedColumn> computedColumns) {
+ super(tableNameConverter, caseSensitive);
+ this.computedColumns = computedColumns;
+ }
+
+ @Override
+ public List<RichCdcMultiplexRecord> extractRecords() {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
+ String type = extractString(FIELD_TYPE);
+ switch (type) {
+ case OP_UPDATE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.UPDATE_BEFORE,
records);
+ processRecord(root.get(FIELD_AFTER), RowKind.UPDATE_AFTER,
records);
+ break;
+ case OP_INSERT:
+ processRecord(root.get(FIELD_AFTER), RowKind.INSERT, records);
+ break;
+ case OP_DELETE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.DELETE, records);
+ break;
+ default:
+ throw new UnsupportedOperationException("Unknown record type:
" + type);
+ }
+ return records;
+ }
+
+ private void processRecord(
+ JsonNode jsonNode, RowKind rowKind, List<RichCdcMultiplexRecord>
records) {
+ LinkedHashMap<String, DataType> paimonFieldTypes = new
LinkedHashMap<>();
+ Map<String, String> rowData = extractRow(jsonNode, paimonFieldTypes);
+ rowData = caseSensitive ? rowData : keyCaseInsensitive(rowData);
+ records.add(createRecord(rowKind, rowData, paimonFieldTypes));
+ }
+
+ private RichCdcMultiplexRecord createRecord(
+ RowKind rowKind,
+ Map<String, String> data,
+ LinkedHashMap<String, DataType> paimonFieldTypes) {
+ return new RichCdcMultiplexRecord(
+ databaseName,
+ tableName,
+ paimonFieldTypes,
+ extractPrimaryKeys(FIELD_PRIMARY_KEYS),
+ new CdcRecord(rowKind, data));
+ }
+
+ @Override
+ public KafkaSchema getKafkaSchema(String record) {
+ try {
+ root = OBJECT_MAPPER.readValue(record, JsonNode.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Error processing JSON: " + record, e);
+ }
+ validateFormat();
+
+ LinkedHashMap<String, DataType> paimonFieldTypes =
extractFieldTypesFromOracleType();
+
+ return new KafkaSchema(
+ databaseName, tableName, paimonFieldTypes,
extractPrimaryKeys(FIELD_PRIMARY_KEYS));
+ }
+
+ @Override
+ protected void validateFormat() {
+ String errorMessageTemplate =
+ "Didn't find '%s' node in json. Please make sure your topic's
format is correct.";
+
+ checkNotNull(root.get(FIELD_TABLE), errorMessageTemplate, FIELD_TABLE);
+ checkNotNull(root.get(FIELD_TYPE), errorMessageTemplate, FIELD_TYPE);
+ checkNotNull(root.get(FIELD_PRIMARY_KEYS), errorMessageTemplate,
FIELD_PRIMARY_KEYS);
+
+ String fieldType = root.get(FIELD_TYPE).asText();
+
+ switch (fieldType) {
+ case OP_UPDATE:
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate,
FIELD_AFTER);
+ checkNotNull(root.get(FIELD_BEFORE), errorMessageTemplate,
FIELD_BEFORE);
+ break;
+ case OP_INSERT:
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate,
FIELD_AFTER);
+ break;
+ case OP_DELETE:
+ checkNotNull(root.get(FIELD_BEFORE), errorMessageTemplate,
FIELD_BEFORE);
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate,
"null");
+ break;
+ }
+ }
+
+ @Override
+ protected String extractString(String key) {
+ if (key.equals(FIELD_TABLE)) {
+ extractDatabaseAndTableNames();
+ return tableName;
+ } else if (key.equals(FIELD_DATABASE)) {
+ extractDatabaseAndTableNames();
+ return databaseName;
+ }
+ return root.get(key) != null ? root.get(key).asText() : null;
+ }
+
+ private void extractDatabaseAndTableNames() {
+ JsonNode tableNode = root.get(FIELD_TABLE);
+ if (tableNode != null) {
+ String[] dbt = tableNode.asText().split("\\.", 2); // Limit split
to 2 parts
+ if (dbt.length == 2) {
+ databaseName = dbt[0];
+ tableName = dbt[1];
+ }
+ }
+ }
+
+ private LinkedHashMap<String, DataType> extractFieldTypesFromOracleType() {
+ LinkedHashMap<String, DataType> fieldTypes = new LinkedHashMap<>();
+
+ JsonNode record = root.get(FIELD_AFTER);
+ Map<String, Object> linkedHashMap =
+ OBJECT_MAPPER.convertValue(
+ record, new TypeReference<LinkedHashMap<String,
Object>>() {});
+ if (linkedHashMap == null) {
+ return new LinkedHashMap<>();
+ }
+
+ Set<String> keySet = linkedHashMap.keySet();
+ String[] columns = keySet.toArray(new String[0]);
+ for (String column : columns) {
+ fieldTypes.put(toFieldName(column), DataTypes.STRING());
+ }
+ return fieldTypes;
+ }
+
+ private Map<String, String> extractRow(
+ JsonNode record, LinkedHashMap<String, DataType> paimonFieldTypes)
{
+ Map<String, String> linkedHashMap =
+ OBJECT_MAPPER.convertValue(
+ record, new TypeReference<LinkedHashMap<String,
String>>() {});
+ if (linkedHashMap == null) {
+ return new HashMap<>();
+ }
+
+ Map<String, String> resultMap = new HashMap<>();
+ for (Map.Entry<String, String> entry : linkedHashMap.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ paimonFieldTypes.put(key, DataTypes.STRING());
+ resultMap.put(key, value);
+ }
+
+ // generate values for computed columns
+ for (ComputedColumn computedColumn : computedColumns) {
+ resultMap.put(
+ computedColumn.columnName(),
+
computedColumn.eval(resultMap.get(computedColumn.fieldReference())));
Review Comment:
Computed column value is evaluated, should pad the paimonFieldTypes?
I also noticed that CanalRecordParser `extractRecords` has the same problem,
please fix it.
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaOggSyncDatabaseActionITCase.java:
##########
@@ -0,0 +1,575 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.kafka;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** IT cases for {@link KafkaSyncDatabaseAction}. */
+public class KafkaOggSyncDatabaseActionITCase extends KafkaActionITCaseBase {
+
+ @Test
+ @Timeout(60)
+ public void testSchemaEvolutionMultiTopic() throws Exception {
+
+ final String topic1 = "schema_evolution_0";
+ final String topic2 = "schema_evolution_1";
+ boolean writeOne = false;
+ int fileCount = 2;
+ List<String> topics = Arrays.asList(topic1, topic2);
+ topics.forEach(
+ topic -> {
+ createTestTopic(topic, 1, 1);
+ });
+
+ // ---------- Write the ogg json into Kafka -------------------
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ topics.get(i),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-1.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+ kafkaConfig.put("topic", String.join(";", topics));
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(2);
+ env.enableCheckpointing(1000);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ Map<String, String> tableConfig = new HashMap<>();
+ tableConfig.put("bucket", String.valueOf(random.nextInt(3) + 1));
+ tableConfig.put("sink.parallelism", String.valueOf(random.nextInt(3) +
1));
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig, warehouse, database,
Collections.emptyMap(), tableConfig);
+ action.build(env);
+ JobClient client = env.executeAsync();
+ waitJobRunning(client);
+
+ testSchemaEvolutionImpl(topics, writeOne, fileCount);
+ }
+
+ @Test
+ @Timeout(60)
+ public void testSchemaEvolutionOneTopic() throws Exception {
+
+ final String topic = "schema_evolution";
+ boolean writeOne = true;
+ int fileCount = 2;
+ List<String> topics = Collections.singletonList(topic);
+ topics.forEach(
+ t -> {
+ createTestTopic(t, 1, 1);
+ });
+
+ // ---------- Write the ogg json into Kafka -------------------
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ topics.get(0),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-1.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+ kafkaConfig.put("topic", String.join(";", topics));
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(2);
+ env.enableCheckpointing(1000);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ Map<String, String> tableConfig = new HashMap<>();
+ tableConfig.put("bucket", String.valueOf(random.nextInt(3) + 1));
+ tableConfig.put("sink.parallelism", String.valueOf(random.nextInt(3) +
1));
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig, warehouse, database,
Collections.emptyMap(), tableConfig);
+ action.build(env);
+ JobClient client = env.executeAsync();
+ waitJobRunning(client);
+
+ testSchemaEvolutionImpl(topics, writeOne, fileCount);
+ }
+
+ private void testSchemaEvolutionImpl(List<String> topics, boolean
writeOne, int fileCount)
+ throws Exception {
+ waitTablesCreated("T1", "T2");
+
+ FileStoreTable table1 = getFileStoreTable("T1");
+ FileStoreTable table2 = getFileStoreTable("T2");
+
+ RowType rowType1 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys1 = Collections.singletonList("id");
+ List<String> expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter,
3.140000104904175]",
+ "+I[102, car battery, 12V car battery,
8.100000381469727]");
+ waitForResult(expected, table1, rowType1, primaryKeys1);
+
+ RowType rowType2 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys2 = Collections.singletonList("id");
+ List<String> expected2 =
+ Arrays.asList(
+ "+I[103, 12-pack drill bits, 12-pack of drill bits
with sizes ranging from #40 to #3, 0.800000011920929]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75]");
+ waitForResult(expected2, table2, rowType2, primaryKeys2);
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ writeOne ? topics.get(0) : topics.get(i),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-2.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ rowType1 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight",
"age"});
+ expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter,
3.140000104904175, NULL]",
+ "+I[102, car battery, 12V car battery,
8.100000381469727, NULL]",
+ "+I[103, 12-pack drill bits, 12-pack of drill bits
with sizes ranging from #40 to #3, 0.800000011920929, 19]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, 25]");
+ waitForResult(expected, table1, rowType1, primaryKeys1);
+
+ rowType2 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight",
"address"});
+ expected =
+ Arrays.asList(
+ "+I[103, 12-pack drill bits, 12-pack of drill bits
with sizes ranging from #40 to #3, 0.800000011920929, Beijing]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75,
Shanghai]");
+ waitForResult(expected, table2, rowType2, primaryKeys2);
+ }
+
+ @Test
+ public void testTopicIsEmpty() {
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig,
+ warehouse,
+ database,
+ Collections.emptyMap(),
+ Collections.emptyMap());
+
+ assertThatThrownBy(() -> action.build(env))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("kafka-conf [topic] must be specified.");
Review Comment:
It's better to use
```
assertThatThrownBy(() -> action.build(env))
.satisfies(
AssertionUtils.anyCauseMatches(....));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
