This is an automated email from the ASF dual-hosted git repository.
renqs pushed a commit to branch release-3.1
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/release-3.1 by this push:
new f476a5b1d [FLINK-35592][cdc] Fix MysqlDebeziumTimeConverter miss
timezone convert to timestamp (#3380)
f476a5b1d is described below
commit f476a5b1dd99fc63962bc75048c0b12957e53da8
Author: ConradJam <[email protected]>
AuthorDate: Thu Jun 13 22:22:09 2024 +0800
[FLINK-35592][cdc] Fix MysqlDebeziumTimeConverter miss timezone convert to
timestamp (#3380)
---
.../converters/MysqlDebeziumTimeConverter.java | 112 ++++++++++++---------
.../MysqlDebeziumTimeConverterITCase.java | 8 +-
.../src/test/resources/ddl/date_convert_test.sql | 3 +-
3 files changed, 73 insertions(+), 50 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/converters/MysqlDebeziumTimeConverter.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/converters/MysqlDebeziumTimeConverter.java
index 493fd682c..1eb98947d 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/converters/MysqlDebeziumTimeConverter.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/converters/MysqlDebeziumTimeConverter.java
@@ -113,54 +113,73 @@ public class MysqlDebeziumTimeConverter
registration.register(
SchemaBuilder.string().name(schemaName).optional(),
value -> {
- log.debug(
- "find schema need to change dateType, field
name:{} ,field type:{} ,field value:{} ,field "
- + "default:{}",
- field.name(),
- columnType,
- value == null ? "null" : value,
- field.hasDefaultValue() ? field.defaultValue() :
"null");
- if (value == null) {
- return convertDateDefaultValue(field);
- }
- switch (columnType.toUpperCase(Locale.ROOT)) {
- case "DATE":
- if (value instanceof Integer) {
- return this.convertToDate(
- columnType,
LocalDate.ofEpochDay((Integer) value));
- }
- return this.convertToDate(columnType, value);
- case "TIME":
- if (value instanceof Long) {
- long l =
- Math.multiplyExact(
- (Long) value,
TimeUnit.MICROSECONDS.toNanos(1));
- return this.convertToTime(columnType,
LocalTime.ofNanoOfDay(l));
- }
- return this.convertToTime(columnType, value);
- case "DATETIME":
- if (value instanceof Long) {
- if (getTimePrecision(field) <= 3) {
- return this.convertToTimestamp(
- columnType,
-
Conversions.toInstantFromMillis((Long) value));
- }
- if (getTimePrecision(field) <= 6) {
- return this.convertToTimestamp(
- columnType,
-
Conversions.toInstantFromMicros((Long) value));
- }
- }
- return this.convertToTimestamp(columnType, value);
- case "TIMESTAMP":
- return
this.convertToTimestampWithTimezone(columnType, value);
- default:
- throw new IllegalArgumentException(
- "Unknown field type " +
columnType.toUpperCase(Locale.ROOT));
+ try {
+ return convertDateObject(field, value, columnType);
+ } catch (Exception e) {
+ printConvertDateErrorClassLogs(field, registration,
value);
+ throw new RuntimeException("MysqlDebeziumConverter
error", e);
}
});
}
+ private void printConvertDateErrorClassLogs(
+ RelationalColumn field,
+ ConverterRegistration<SchemaBuilder> registration,
+ Object value) {
+ boolean useDefaultValueConvert = (value == null);
+ String fieldName = field.name();
+ String fieldType = field.typeName().toUpperCase();
+ String defaultValue = "null";
+ if (field.hasDefaultValue()) {
+ if (field.defaultValue() != null) {
+ defaultValue = field.defaultValue().toString();
+ }
+ }
+ log.warn(
+ "find schema need to change dateType, field name:||{}|| field
type:||{}|| is use default "
+ + "convert:||{}|| field default value:||{}|| field
charge value fail",
+ fieldName,
+ fieldType,
+ useDefaultValueConvert,
+ defaultValue);
+ }
+
+ private Object convertDateObject(RelationalColumn field, Object value,
String columnType) {
+ if (value == null) {
+ return convertDateDefaultValue(field);
+ }
+ switch (columnType.toUpperCase(Locale.ROOT)) {
+ case "DATE":
+ if (value instanceof Integer) {
+ return this.convertToDate(columnType,
LocalDate.ofEpochDay((Integer) value));
+ }
+ return this.convertToDate(columnType, value);
+ case "TIME":
+ if (value instanceof Long) {
+ long l = Math.multiplyExact((Long) value,
TimeUnit.MICROSECONDS.toNanos(1));
+ return this.convertToTime(columnType,
LocalTime.ofNanoOfDay(l));
+ }
+ return this.convertToTime(columnType, value);
+ case "DATETIME":
+ if (value instanceof Long) {
+ if (getTimePrecision(field) <= 3) {
+ return this.convertToTimestamp(
+ columnType,
Conversions.toInstantFromMillis((Long) value));
+ }
+ if (getTimePrecision(field) <= 6) {
+ return this.convertToTimestamp(
+ columnType,
Conversions.toInstantFromMicros((Long) value));
+ }
+ }
+ return this.convertToTimestamp(columnType, value);
+ case "TIMESTAMP":
+ return this.convertToTimestampWithTimezone(columnType, value);
+ default:
+ throw new IllegalArgumentException(
+ "Unknown field type " +
columnType.toUpperCase(Locale.ROOT));
+ }
+ }
+
private Object convertToTimestampWithTimezone(String columnType, Object
timestamp) {
// In snapshot mode, debezium produces a java.sql.Timestamp object for
the TIMESTAMPTZ type.
// Conceptually, a timestamp with timezone is an Instant. But
t.toInstant() actually
@@ -174,11 +193,12 @@ public class MysqlDebeziumTimeConverter
ZonedDateTime zonedDateTime = value.toInstant().atZone(zoneId);
return ConvertTimeBceUtil.resolveEra(value,
zonedDateTime.format(timestampFormatter));
} else if (timestamp instanceof OffsetDateTime) {
- OffsetDateTime value = (OffsetDateTime) timestamp;
+ OffsetDateTime value =
+ ((OffsetDateTime)
timestamp).toInstant().atZone(zoneId).toOffsetDateTime();
return ConvertTimeBceUtil.resolveEra(
value.toLocalDate(), value.format(timestampFormatter));
} else if (timestamp instanceof ZonedDateTime) {
- ZonedDateTime zonedDateTime = (ZonedDateTime) timestamp;
+ ZonedDateTime zonedDateTime = ((ZonedDateTime)
timestamp).toInstant().atZone(zoneId);
return ConvertTimeBceUtil.resolveEra(
zonedDateTime.toLocalDate(),
zonedDateTime.format(timestampFormatter));
} else if (timestamp instanceof Instant) {
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/converters/MysqlDebeziumTimeConverterITCase.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/converters/MysqlDebeziumTimeConverterITCase.java
index d14f31f97..8adb92454 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/converters/MysqlDebeziumTimeConverterITCase.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/converters/MysqlDebeziumTimeConverterITCase.java
@@ -154,7 +154,7 @@ public class MysqlDebeziumTimeConverterITCase {
private void validTimestampValue(List<String> result) throws
JsonProcessingException {
ObjectMapper mapper = new ObjectMapper();
- String[] timestampValues = new String[] {"14:23:00", "00:00:00",
"00:00:00"};
+ String[] timestampValues = new String[] {"14:23:00", "00:00:00",
"00:00:00", "15:04:00"};
for (String after : result) {
JsonNode jsonNode = mapper.readTree(after);
Assert.assertEquals(
@@ -232,7 +232,8 @@ public class MysqlDebeziumTimeConverterITCase {
new String[] {
"+I[1, 14:23:00, 2023-04-01 14:24:00, 2023-04-01,
14:25:00]",
"+I[3, 00:00:00, null, null, 00:01:20]",
- "+I[2, 00:00:00, null, null, 00:00:00]"
+ "+I[2, 00:00:00, null, null, 00:00:00]",
+ "+I[4, 15:04:00, null, null, 00:01:10]"
};
List<String> expectedSnapshotData = new
ArrayList<>(Arrays.asList(snapshotForSingleTable));
@@ -283,7 +284,8 @@ public class MysqlDebeziumTimeConverterITCase {
+ "binlog_format = row\n"
+ "log_bin = mysql-bin\n"
+ "server-id = 223344\n"
- + "binlog_row_image = FULL\n";
+ + "binlog_row_image = FULL\n"
+ + "sql_mode =
ALLOW_INVALID_DATES,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION\n";
String timezoneConf = "default-time_zone = '" + timezone + "'\n";
Files.write(
cnf,
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/resources/ddl/date_convert_test.sql
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/resources/ddl/date_convert_test.sql
index 262c1ceb1..ed9aadfdc 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/resources/ddl/date_convert_test.sql
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/resources/ddl/date_convert_test.sql
@@ -33,4 +33,5 @@ INSERT INTO date_convert_test (id,test_timestamp,
test_datetime, test_date, test
VALUES
(1,'2023-04-01 14:23:00', '2023-04-01 14:24:00', '2023-04-01', '14:25:00'),
(2,'2024-04-23 00:00:00', DEFAULT, NULL ,'00:00:00'),
-(3,'2024-04-23 00:00:00', DEFAULT, NULL ,120);
\ No newline at end of file
+(3,'2024-04-23 00:00:00', DEFAULT, NULL ,120),
+(4,20240612150400, DEFAULT, NULL ,110);
\ No newline at end of file
