github-actions[bot] commented on code in PR #41331:
URL: https://github.com/apache/beam/pull/41331#discussion_r4572723621
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java:
##########
@@ -118,12 +118,15 @@ public class SourceRecordJson {
private static final String SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME =
"io.debezium.connector.common.TransactionMetadataKey";
+ public static final String SCHEMA_CHANGE_EVENT_KEY_NAME =
Review Comment:
**🟢 Low**: `SCHEMA_CHANGE_EVENT_KEY_NAME` is an internal implementation
detail of `SourceRecordJson` (just like
`SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME` above it) and should be `private`
to avoid expanding the public API surface unnecessarily.
```suggestion
private static final String SCHEMA_CHANGE_EVENT_KEY_NAME =
```
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java:
##########
@@ -592,6 +716,10 @@ private static boolean
isTransactionMetadataEvent(@Nullable Schema keySchema) {
return keySchema != null &&
SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME.equals(keySchema.name());
}
+ private static boolean isSchemaChangeEventKeyName(@Nullable Schema
keySchema) {
+ return keySchema != null &&
SCHEMA_CHANGE_EVENT_KEY_NAME.equals(keySchema.name());
Review Comment:
**🟡 Medium**: In Debezium's `SchemaFactory`, schema change key schemas are
named `io.debezium.connector.<connector>.SchemaChangeKey` (e.g., `mysql`,
`sqlserver`, `oracle`, `db2`). Since `DebeziumIO` supports multiple relational
connectors and `ChangeSchemaEvent` already maps `schemaName`, consider matching
any connector's `SchemaChangeKey` (e.g., checking prefix
`"io.debezium.connector."` and suffix `".SchemaChangeKey"`) rather than only
MySQL.
##########
sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java:
##########
@@ -367,4 +371,91 @@ public void testDebeziumIOMySqlFullMetadata() {
p.run().waitUntilFinish();
}
+
+ /**
+ * Debezium - MySQL connector Test with {@code include.schema.changes}
enabled.
+ *
+ * <p>Tests that the snapshot's CREATE schema change event for {@code
inventory.customers} carries
+ * the table's columns, each with every field of Debezium's column schema.
+ */
+ @Test
+ public void testDebeziumIOMySqlSchemaChangeEventColumns() {
+
+ String kafkaBootstrapServers = KAFKA_CONTAINER.getBootstrapServers();
+ String schemaHistoryTopic = "mysql-schema-history-schema-change-" +
System.nanoTime();
+
+ String host = MY_SQL_CONTAINER.getHost();
+ String port = MY_SQL_CONTAINER.getMappedPort(3306).toString();
+
+ PipelineOptions options = PipelineOptionsFactory.create();
+ Pipeline p = Pipeline.create(options);
+ PCollection<String> results =
+ p.apply(
+ DebeziumIO.readAsJson()
+ .withConnectorConfiguration(
+ DebeziumIO.ConnectorConfiguration.create()
+ .withUsername("debezium")
+ .withPassword("dbz")
+ .withConnectorClass(MySqlConnector.class)
+ .withHostName(host)
+ .withPort(port)
+ .withConnectionProperty("database.server.id", "184056")
+ .withConnectionProperty("database.include.list",
"inventory")
+ .withConnectionProperty("table.include.list",
"inventory.customers")
+ .withConnectionProperty("include.schema.changes",
"true")
+ .withConnectionProperty(
+ "schema.history.internal.kafka.bootstrap.servers",
+ kafkaBootstrapServers)
+ .withConnectionProperty(
+ "schema.history.internal.kafka.topic",
schemaHistoryTopic)
+ .withConnectionProperty(
+ "schema.history.internal",
+
"io.debezium.storage.kafka.history.KafkaSchemaHistory")
+ .withConnectionProperty(
+
"schema.history.internal.store.only.captured.tables.ddl", "true"))
+ // The snapshot emits its schema change events before the
customers rows.
+ .withMaxNumberOfRecords(10)
+ .withMaxTimeToRun(Duration.ofMinutes(2).toMillis()));
+
+ // Every field of Debezium's column schema, in schema order, nulls
included.
+ String expectedFirstNameColumn =
+
"{\"name\":\"first_name\",\"jdbcType\":12,\"nativeType\":null,\"typeName\":\"VARCHAR\","
+ +
"\"typeExpression\":\"VARCHAR\",\"charsetName\":\"utf8mb4\",\"length\":255,"
+ +
"\"scale\":null,\"position\":2,\"optional\":false,\"autoIncremented\":false,"
+ +
"\"generated\":false,\"comment\":null,\"defaultValueExpression\":null,"
+ + "\"enumValues\":null}";
+
+ PAssert.that(results)
+ .satisfies(
+ (Iterable<String> res) -> {
+ // Find the snapshot's CREATE event for the customers table.
+ String customersId = "\"inventory\".\"customers\"";
+ JsonObject table = null;
+ for (String record : res) {
+ JsonObject event =
JsonParser.parseString(record).getAsJsonObject();
Review Comment:
**🟢 Low**: Because `SourceRecordJson` configures `Gson` with
`.serializeNulls()`, `event.has("tableChanges")` will still return `true` if a
schema change event has `"tableChanges": null`, causing
`event.getAsJsonArray("tableChanges")` to throw an `IllegalStateException`.
Also consider asserting `assertNotNull(table)` after the loop before
dereferencing `table.get("primaryKeyColumnNames")` on line 447 so test failures
produce a clear assertion error rather than a `NullPointerException`.
```suggestion
if (!event.has("tableChanges") ||
event.get("tableChanges").isJsonNull()) {
```
--
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]
