| Option | +Required | +Default | +Type | +Description | +
|---|---|---|---|---|
| connector | +required | +(none) | +String | +Specify what connector to use, here should be ‘vitess-cdc’. |
+
| hostname | +required | +(none) | +String | +IP address or hostname of the Vitess database server (VTGate). | +
| keyspace | +required | +(none) | +String | +The name of the keyspace from which to stream the changes. | +
| username | +optional | +(none) | +String | +An optional username of the Vitess database server (VTGate). If not configured, unauthenticated VTGate gRPC is used. | +
| password | +optional | +(none) | +String | +An optional password of the Vitess database server (VTGate). If not configured, unauthenticated VTGate gRPC is used. | +
| table-name | +required | +(none) | +String | +Table name of the MySQL database to monitor. | +
| port | +optional | +15991 | +Integer | +Integer port number of the VTCtld server. | +
| vtctld.host | +required | +(none) | +String | +IP address or hostname of the VTCtld server. | +
| vtctld.port | +optional | +15999 | +Integer | +Integer port number of the VTCtld server. | +
| vtctld.user | +optional | +(none) | +String | +An optional username of the VTCtld server. If not configured, unauthenticated VTCtld gRPC is used. | +
| vtctld.password | +optional | +(none) | +String | +An optional password of the VTCtld server. If not configured, unauthenticated VTCtld gRPC is used. | +
| tablet.type | +optional | +RDONLY | +String | +The type of Tablet (hence MySQL) from which to stream the changes: MASTER represents streaming from the master MySQL instance REPLICA represents streaming from the replica slave MySQL instance RDONLY represents streaming from the read-only slave MySQL instance. | +
extends EventDispatcher Note https://issues.redhat.com/browse/DBZ-3238 has fixed this issue, please remove
- * this method once we bump Debezium version to 1.6
- */
- private Object readField(ResultSet rs, int fieldNo, Column actualColumn, Table actualTable)
- throws SQLException {
- if (actualColumn.jdbcType() == Types.TIME) {
- return readTimeField(rs, fieldNo);
- } else if (actualColumn.jdbcType() == Types.DATE) {
- return readDateField(rs, fieldNo, actualColumn, actualTable);
- }
- // This is for DATETIME columns (a logical date + time without time zone)
- // by reading them with a calendar based on the default time zone, we make sure that the
- // value
- // is constructed correctly using the database's (or connection's) time zone
- else if (actualColumn.jdbcType() == Types.TIMESTAMP) {
- return readTimestampField(rs, fieldNo, actualColumn, actualTable);
- }
- // JDBC's rs.GetObject() will return a Boolean for all TINYINT(1) columns.
- // TINYINT columns are reprtoed as SMALLINT by JDBC driver
- else if (actualColumn.jdbcType() == Types.TINYINT
- || actualColumn.jdbcType() == Types.SMALLINT) {
- // It seems that rs.wasNull() returns false when default value is set and NULL is
- // inserted
- // We thus need to use getObject() to identify if the value was provided and if yes
- // then
- // read it again to get correct scale
- return rs.getObject(fieldNo) == null ? null : rs.getInt(fieldNo);
- }
- // DBZ-2673
- // It is necessary to check the type names as types like ENUM and SET are
- // also reported as JDBC type char
- else if ("CHAR".equals(actualColumn.typeName())
- || "VARCHAR".equals(actualColumn.typeName())
- || "TEXT".equals(actualColumn.typeName())) {
- return rs.getBytes(fieldNo);
- } else {
- return rs.getObject(fieldNo);
- }
- }
-
- /**
- * As MySQL connector/J implementation is broken for MySQL type "TIME" we have to use a
- * binary-ish workaround. https://issues.jboss.org/browse/DBZ-342
- */
- private Object readTimeField(ResultSet rs, int fieldNo) throws SQLException {
- Blob b = rs.getBlob(fieldNo);
- if (b == null) {
- return null; // Don't continue parsing time field if it is null
- }
-
- try {
- return MySqlValueConverters.stringToDuration(
- new String(b.getBytes(1, (int) (b.length())), "UTF-8"));
- } catch (UnsupportedEncodingException e) {
- LOG.error("Could not read MySQL TIME value as UTF-8");
- throw new RuntimeException(e);
- }
- }
-
- /**
- * In non-string mode the date field can contain zero in any of the date part which we need
- * to handle as all-zero.
- */
- private Object readDateField(ResultSet rs, int fieldNo, Column column, Table table)
- throws SQLException {
- Blob b = rs.getBlob(fieldNo);
- if (b == null) {
- return null; // Don't continue parsing date field if it is null
- }
-
- try {
- return MySqlValueConverters.stringToLocalDate(
- new String(b.getBytes(1, (int) (b.length())), "UTF-8"), column, table);
- } catch (UnsupportedEncodingException e) {
- LOG.error("Could not read MySQL TIME value as UTF-8");
- throw new RuntimeException(e);
- }
- }
-
- /**
- * In non-string mode the time field can contain zero in any of the date part which we need
- * to handle as all-zero.
- */
- private Object readTimestampField(ResultSet rs, int fieldNo, Column column, Table table)
- throws SQLException {
- Blob b = rs.getBlob(fieldNo);
- if (b == null) {
- return null; // Don't continue parsing timestamp field if it is null
- }
-
- try {
- return MySqlValueConverters.containsZeroValuesInDatePart(
- (new String(b.getBytes(1, (int) (b.length())), "UTF-8")),
- column,
- table)
- ? null
- : rs.getTimestamp(fieldNo, Calendar.getInstance());
- } catch (UnsupportedEncodingException e) {
- LOG.error("Could not read MySQL TIME value as UTF-8");
- throw new RuntimeException(e);
- }
- }
}
/**
* {@link ChangeEventSource.ChangeEventSourceContext} implementation that keeps low/high
* watermark for each {@link SnapshotSplit}.
*/
- public class SnapshotSplitChangeEventSourceContext
+ public static class SnapshotSplitChangeEventSourceContext
implements ChangeEventSource.ChangeEventSourceContext {
private BinlogOffset lowWatermark;
diff --git a/flink-cdc-base/src/test/java/com/ververica/cdc/connectors/base/experimental/fetch/MySqlSourceFetchTaskContext.java b/flink-cdc-base/src/test/java/com/ververica/cdc/connectors/base/experimental/fetch/MySqlSourceFetchTaskContext.java
index 49b4a33c4a8..cf1eecc322b 100644
--- a/flink-cdc-base/src/test/java/com/ververica/cdc/connectors/base/experimental/fetch/MySqlSourceFetchTaskContext.java
+++ b/flink-cdc-base/src/test/java/com/ververica/cdc/connectors/base/experimental/fetch/MySqlSourceFetchTaskContext.java
@@ -37,6 +37,7 @@
import io.debezium.connector.mysql.MySqlDatabaseSchema;
import io.debezium.connector.mysql.MySqlErrorHandler;
import io.debezium.connector.mysql.MySqlOffsetContext;
+import io.debezium.connector.mysql.MySqlPartition;
import io.debezium.connector.mysql.MySqlStreamingChangeEventSourceMetrics;
import io.debezium.connector.mysql.MySqlTaskContext;
import io.debezium.connector.mysql.MySqlTopicSelector;
@@ -46,6 +47,7 @@
import io.debezium.pipeline.metrics.SnapshotChangeEventSourceMetrics;
import io.debezium.pipeline.source.spi.EventMetadataProvider;
import io.debezium.pipeline.spi.OffsetContext;
+import io.debezium.pipeline.spi.Offsets;
import io.debezium.relational.Table;
import io.debezium.relational.TableId;
import io.debezium.relational.Tables;
@@ -75,10 +77,11 @@ public class MySqlSourceFetchTaskContext extends JdbcSourceFetchTaskContext {
private MySqlDatabaseSchema databaseSchema;
private MySqlTaskContextImpl taskContext;
private MySqlOffsetContext offsetContext;
- private SnapshotChangeEventSourceMetrics snapshotChangeEventSourceMetrics;
+ private SnapshotChangeEventSourceMetrics Added JMX_METRICS_ENABLED option.
+ */
+public abstract class HistorizedRelationalDatabaseConnectorConfig
+ extends RelationalDatabaseConnectorConfig {
+
+ protected static final int DEFAULT_SNAPSHOT_FETCH_SIZE = 2_000;
+
+ private boolean useCatalogBeforeSchema;
+ private final String logicalName;
+ private final Class connectorClass;
+ private final boolean multiPartitionMode;
+
+ /**
+ * The database history class is hidden in the {@link #configDef()} since that is designed to
+ * work with a user interface, and in these situations using Kafka is the only way to go.
+ */
+ public static final Field DATABASE_HISTORY =
+ Field.create("database.history")
+ .withDisplayName("Database history class")
+ .withType(Type.CLASS)
+ .withWidth(Width.LONG)
+ .withImportance(Importance.LOW)
+ .withInvisibleRecommender()
+ .withDescription(
+ "The name of the DatabaseHistory class that should be used to store and recover database schema changes. "
+ + "The configuration properties for the history are prefixed with the '"
+ + DatabaseHistory.CONFIGURATION_FIELD_PREFIX_STRING
+ + "' string.")
+ .withDefault(KafkaDatabaseHistory.class.getName());
+
+ public static final Field JMX_METRICS_ENABLED =
+ Field.create(DatabaseHistory.CONFIGURATION_FIELD_PREFIX_STRING + "metrics.enabled")
+ .withDisplayName("Skip DDL statements that cannot be parsed")
+ .withType(Type.BOOLEAN)
+ .withImportance(Importance.LOW)
+ .withDescription("Whether to enable JMX history metrics")
+ .withDefault(false);
+
+ protected static final ConfigDefinition CONFIG_DEFINITION =
+ RelationalDatabaseConnectorConfig.CONFIG_DEFINITION
+ .edit()
+ .history(
+ DATABASE_HISTORY,
+ DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS,
+ DatabaseHistory.STORE_ONLY_MONITORED_TABLES_DDL,
+ DatabaseHistory.STORE_ONLY_CAPTURED_TABLES_DDL,
+ KafkaDatabaseHistory.BOOTSTRAP_SERVERS,
+ KafkaDatabaseHistory.TOPIC,
+ KafkaDatabaseHistory.RECOVERY_POLL_ATTEMPTS,
+ KafkaDatabaseHistory.RECOVERY_POLL_INTERVAL_MS,
+ KafkaDatabaseHistory.KAFKA_QUERY_TIMEOUT_MS)
+ .create();
+
+ protected HistorizedRelationalDatabaseConnectorConfig(
+ Class connectorClass,
+ Configuration config,
+ String logicalName,
+ TableFilter systemTablesFilter,
+ boolean useCatalogBeforeSchema,
+ int defaultSnapshotFetchSize,
+ ColumnFilterMode columnFilterMode,
+ boolean multiPartitionMode) {
+ super(
+ config,
+ logicalName,
+ systemTablesFilter,
+ TableId::toString,
+ defaultSnapshotFetchSize,
+ columnFilterMode);
+ this.useCatalogBeforeSchema = useCatalogBeforeSchema;
+ this.logicalName = logicalName;
+ this.connectorClass = connectorClass;
+ this.multiPartitionMode = multiPartitionMode;
+ }
+
+ protected HistorizedRelationalDatabaseConnectorConfig(
+ Class connectorClass,
+ Configuration config,
+ String logicalName,
+ TableFilter systemTablesFilter,
+ TableIdToStringMapper tableIdMapper,
+ boolean useCatalogBeforeSchema,
+ ColumnFilterMode columnFilterMode,
+ boolean multiPartitionMode) {
+ super(
+ config,
+ logicalName,
+ systemTablesFilter,
+ tableIdMapper,
+ DEFAULT_SNAPSHOT_FETCH_SIZE,
+ columnFilterMode);
+ this.useCatalogBeforeSchema = useCatalogBeforeSchema;
+ this.logicalName = logicalName;
+ this.connectorClass = connectorClass;
+ this.multiPartitionMode = multiPartitionMode;
+ }
+
+ /** Returns a configured (but not yet started) instance of the database history. */
+ public DatabaseHistory getDatabaseHistory() {
+ Configuration config = getConfig();
+
+ DatabaseHistory databaseHistory =
+ config.getInstance(
+ HistorizedRelationalDatabaseConnectorConfig.DATABASE_HISTORY,
+ DatabaseHistory.class);
+ if (databaseHistory == null) {
+ throw new ConnectException(
+ "Unable to instantiate the database history class "
+ + config.getString(
+ HistorizedRelationalDatabaseConnectorConfig.DATABASE_HISTORY));
+ }
+
+ // Do not remove the prefix from the subset of config properties ...
+ Configuration dbHistoryConfig =
+ config.subset(DatabaseHistory.CONFIGURATION_FIELD_PREFIX_STRING, false)
+ .edit()
+ .withDefault(DatabaseHistory.NAME, getLogicalName() + "-dbhistory")
+ .withDefault(
+ KafkaDatabaseHistory.INTERNAL_CONNECTOR_CLASS,
+ connectorClass.getName())
+ .withDefault(KafkaDatabaseHistory.INTERNAL_CONNECTOR_ID, logicalName)
+ .build();
+
+ DatabaseHistoryListener listener =
+ config.getBoolean(JMX_METRICS_ENABLED)
+ ? new DatabaseHistoryMetrics(this, multiPartitionMode)
+ : DatabaseHistoryListener.NOOP;
+
+ HistoryRecordComparator historyComparator = getHistoryRecordComparator();
+ databaseHistory.configure(
+ dbHistoryConfig, historyComparator, listener, useCatalogBeforeSchema); // validates
+
+ return databaseHistory;
+ }
+
+ public boolean useCatalogBeforeSchema() {
+ return useCatalogBeforeSchema;
+ }
+
+ /**
+ * Returns a comparator to be used when recovering records from the schema history, making sure
+ * no history entries newer than the offset we resume from are recovered (which could happen
+ * when restarting a connector after history records have been persisted but no new offset has
+ * been committed yet).
+ */
+ protected abstract HistoryRecordComparator getHistoryRecordComparator();
+}
diff --git a/flink-connector-mongodb-cdc/pom.xml b/flink-connector-mongodb-cdc/pom.xml
index b1b1732492e..9285b05481a 100644
--- a/flink-connector-mongodb-cdc/pom.xml
+++ b/flink-connector-mongodb-cdc/pom.xml
@@ -159,6 +159,12 @@ under the License.
Line 71: Add field {@code urlPattern} in {@link MySqlConnection} and remove old pattern.
+ * Line 75: Add field {@code urlPattern} in {@link MySqlConnection} and remove old pattern.
*
- * Line 84: Init {@code urlPattern} using the url pattern from {@link
+ * Line 92: Init {@code urlPattern} using the url pattern from {@link
* MySqlConnectionConfiguration}.
*
- * Line 552: Generate the connection string by the new field {@code urlPattern}.
+ * Line 544: Generate the connection string by the new field {@code urlPattern}.
*
- * Line 566 ~ 577: Add new constant and field {@code urlPattern} to {@link
+ * Line 569 ~ 574: Add new constant and field {@code urlPattern} to {@link
* MySqlConnectionConfiguration}.
*
- * Line 622 ~ 625: Init new field {@code urlPattern} in {@link MySqlConnectionConfiguration}.
+ * Line 625: Init new field {@code urlPattern} in {@link MySqlConnectionConfiguration}.
*
- * Line 686 ~ 716: Add some methods helping to generate the url pattern and add default values.
+ * Line 715 ~ 741: Add some methods helping to generate the url pattern and add default values.
*/
public class MySqlConnection extends JdbcConnection {
@@ -64,10 +65,12 @@ public class MySqlConnection extends JdbcConnection {
"SHOW VARIABLES WHERE Variable_name IN ('character_set_server','collation_server')";
private static final String SQL_SHOW_SESSION_VARIABLE_SSL_VERSION =
"SHOW SESSION STATUS LIKE 'Ssl_version'";
+ private static final String QUOTED_CHARACTER = "`";
private final Map Line 87, Line 114~129: trim the default string value when the type is a numeric type.
- */
-@Immutable
-public class MySqlDefaultValueConverter {
-
- private static final Pattern EPOCH_EQUIVALENT_TIMESTAMP =
- Pattern.compile(
- "(\\d{4}-\\d{2}-00|\\d{4}-00-\\d{2}|0000-\\d{2}-\\d{2}) (00:00:00(\\.\\d{1,6})?)");
-
- private static final Pattern EPOCH_EQUIVALENT_DATE =
- Pattern.compile("\\d{4}-\\d{2}-00|\\d{4}-00-\\d{2}|0000-\\d{2}-\\d{2}");
-
- private static final String EPOCH_TIMESTAMP = "1970-01-01 00:00:00";
-
- private static final String EPOCH_DATE = "1970-01-01";
-
- private static final DateTimeFormatter ISO_LOCAL_DATE_WITH_OPTIONAL_TIME =
- new DateTimeFormatterBuilder()
- .append(DateTimeFormatter.ISO_LOCAL_DATE)
- .optionalStart()
- .appendLiteral(" ")
- .append(DateTimeFormatter.ISO_LOCAL_TIME)
- .optionalEnd()
- .toFormatter();
-
- private final MySqlValueConverters converters;
-
- public MySqlDefaultValueConverter(MySqlValueConverters converters) {
- this.converters = converters;
- }
-
- /**
- * Converts a default value from the expected format to a logical object acceptable by the main
- * JDBC converter.
- *
- * @param column column definition
- * @param value string formatted default value
- * @return value converted to a Java type
- */
- public Object convert(Column column, String value) {
- if (value == null) {
- return value;
- }
-
- // boolean is also INT(1) or TINYINT(1)
- if ("TINYINT".equals(column.typeName()) || "INT".equals(column.typeName())) {
- if ("true".equalsIgnoreCase(value) || "false".equalsIgnoreCase(value)) {
- return convertToBoolean(value);
- }
- }
-
- value = trimValueIfNeed(column, value);
- switch (column.jdbcType()) {
- case Types.DATE:
- return convertToLocalDate(column, value);
- case Types.TIMESTAMP:
- return convertToLocalDateTime(column, value);
- case Types.TIMESTAMP_WITH_TIMEZONE:
- return convertToTimestamp(column, value);
- case Types.TIME:
- return convertToDuration(column, value);
- case Types.BOOLEAN:
- return convertToBoolean(value);
- case Types.BIT:
- return convertToBits(column, value);
-
- case Types.NUMERIC:
- case Types.DECIMAL:
- return convertToDecimal(column, value);
-
- case Types.FLOAT:
- case Types.DOUBLE:
- case Types.REAL:
- return convertToDouble(value);
- }
- return value;
- }
-
- private String trimValueIfNeed(Column column, String value) {
- int jdbcType = column.jdbcType();
- if (jdbcType == Types.DECIMAL
- || jdbcType == Types.INTEGER
- || jdbcType == Types.TINYINT
- || jdbcType == Types.BIGINT
- || jdbcType == Types.SMALLINT
- || jdbcType == Types.DOUBLE
- || jdbcType == Types.FLOAT
- || jdbcType == Types.BOOLEAN
- || jdbcType == Types.NUMERIC
- || jdbcType == Types.REAL) {
- return value.trim();
- }
- return value;
- }
-
- /**
- * Converts a string object for an object type of {@link LocalDate} or {@link LocalDateTime} in
- * case of MySql Date type. If the column definition allows null and default value is
- * 0000-00-00, we need return null; else 0000-00-00 will be replaced with 1970-01-01;
- *
- * @param column the column definition describing the {@code data} value; never null
- * @param value the string object to be converted into a {@link LocalDate} type or {@link
- * LocalDateTime} in case of MySql Date type;
- * @return the converted value;
- */
- private Object convertToLocalDate(Column column, String value) {
- final boolean zero =
- EPOCH_EQUIVALENT_DATE.matcher(value).matches()
- || EPOCH_EQUIVALENT_TIMESTAMP.matcher(value).matches()
- || "0".equals(value);
-
- if (zero && column.isOptional()) {
- return null;
- }
- if (zero) {
- value = EPOCH_DATE;
- }
- return LocalDate.from(ISO_LOCAL_DATE_WITH_OPTIONAL_TIME.parse(value));
- }
-
- /**
- * Converts a string object for an object type of {@link LocalDateTime}. If the column
- * definition allows null and default value is 0000-00-00 00:00:00, we need return null, else
- * 0000-00-00 00:00:00 will be replaced with 1970-01-01 00:00:00;
- *
- * @param column the column definition describing the {@code data} value; never null
- * @param value the string object to be converted into a {@link LocalDateTime} type;
- * @return the converted value;
- */
- private Object convertToLocalDateTime(Column column, String value) {
- final boolean matches =
- EPOCH_EQUIVALENT_TIMESTAMP.matcher(value).matches() || "0".equals(value);
- if (matches) {
- if (column.isOptional()) {
- return null;
- }
-
- value = EPOCH_TIMESTAMP;
- }
-
- return LocalDateTime.from(timestampFormat(column.length()).parse(value));
- }
-
- /**
- * Converts a string object for an object type of {@link Timestamp}. If the column definition
- * allows null and default value is 0000-00-00 00:00:00, we need return null, else 0000-00-00
- * 00:00:00 will be replaced with 1970-01-01 00:00:00;
- *
- * @param column the column definition describing the {@code data} value; never null
- * @param value the string object to be converted into a {@link Timestamp} type;
- * @return the converted value;
- */
- private Object convertToTimestamp(Column column, String value) {
- final boolean matches =
- EPOCH_EQUIVALENT_TIMESTAMP.matcher(value).matches()
- || "0".equals(value)
- || EPOCH_TIMESTAMP.equals(value);
- if (matches) {
- if (column.isOptional()) {
- return null;
- }
-
- return Timestamp.from(Instant.EPOCH);
- }
- value = cleanTimestamp(value);
- return Timestamp.valueOf(value).toInstant().atZone(ZoneId.systemDefault());
- }
-
- /**
- * Converts a string object for an object type of {@link Duration}.
- *
- * @param column the column definition describing the {@code data} value; never null
- * @param value the string object to be converted into a {@link Duration} type;
- * @return the converted value;
- */
- private Object convertToDuration(Column column, String value) {
- return MySqlValueConverters.stringToDuration(value);
- }
-
- /**
- * Converts a string object for an expected JDBC type of {@link Types#DOUBLE}.
- *
- * @param value the string object to be converted into a {@link Types#DOUBLE} type;
- * @return the converted value;
- */
- private Object convertToDouble(String value) {
- return Double.parseDouble(value);
- }
-
- /**
- * Converts a string object for an expected JDBC type of {@link Types#DECIMAL}.
- *
- * @param column the column definition describing the {@code data} value; never null
- * @param value the string object to be converted into a {@link Types#DECIMAL} type;
- * @return the converted value;
- */
- private Object convertToDecimal(Column column, String value) {
- return column.scale().isPresent()
- ? new BigDecimal(value).setScale(column.scale().get(), RoundingMode.HALF_UP)
- : new BigDecimal(value);
- }
-
- /**
- * Converts a string object for an expected JDBC type of {@link Types#BIT}.
- *
- * @param column the column definition describing the {@code data} value; never null
- * @param value the string object to be converted into a {@link Types#BIT} type;
- * @return the converted value;
- */
- private Object convertToBits(Column column, String value) {
- if (column.length() > 1) {
- return convertToBits(value);
- }
- return convertToBit(value);
- }
-
- private Object convertToBit(String value) {
- try {
- return Short.parseShort(value) != 0;
- } catch (NumberFormatException ignore) {
- return Boolean.parseBoolean(value);
- }
- }
-
- private Object convertToBits(String value) {
- int nums = value.length() / Byte.SIZE + (value.length() % Byte.SIZE == 0 ? 0 : 1);
- byte[] bytes = new byte[nums];
- for (int i = 0; i < nums; i++) {
- int s = value.length() - Byte.SIZE < 0 ? 0 : value.length() - Byte.SIZE;
- int e = value.length();
- bytes[nums - i - 1] = (byte) Integer.parseInt(value.substring(s, e), 2);
- value = value.substring(0, s);
- }
- return bytes;
- }
-
- /**
- * Converts a string object for an expected JDBC type of {@link Types#BOOLEAN}.
- *
- * @param value the string object to be converted into a {@link Types#BOOLEAN} type;
- * @return the converted value;
- */
- private Object convertToBoolean(String value) {
- try {
- return Integer.parseInt(value) != 0;
- } catch (NumberFormatException ignore) {
- return Boolean.parseBoolean(value);
- }
- }
-
- private DateTimeFormatter timestampFormat(int length) {
- final DateTimeFormatterBuilder dtf =
- new DateTimeFormatterBuilder()
- .appendPattern("yyyy-MM-dd")
- .optionalStart()
- .appendLiteral(" ")
- .append(DateTimeFormatter.ISO_LOCAL_TIME)
- .optionalEnd()
- .parseDefaulting(ChronoField.HOUR_OF_DAY, 0)
- .parseDefaulting(ChronoField.MINUTE_OF_HOUR, 0)
- .parseDefaulting(ChronoField.SECOND_OF_MINUTE, 0);
- if (length > 0) {
- dtf.appendFraction(ChronoField.MICRO_OF_SECOND, 0, length, true);
- }
- return dtf.toFormatter();
- }
-
- /**
- * Clean input timestamp to yyyy-mm-dd hh:mm:ss[.fffffffff] format.
- *
- * @param s input timestamp
- * @return cleaned timestamp
- */
- private String cleanTimestamp(String s) {
- if (s == null) {
- throw new java.lang.IllegalArgumentException("null string");
- }
-
- s = s.trim();
-
- // clean first dash
- s = replaceFirstNonNumericSubstring(s, 0, '-');
- // clean second dash
- s = replaceFirstNonNumericSubstring(s, s.indexOf('-') + 1, '-');
- // clean dividing space
- s = replaceFirstNonNumericSubstring(s, s.indexOf('-', s.indexOf('-') + 1) + 1, ' ');
- if (s.indexOf(' ') != -1) {
- // clean first colon
- s = replaceFirstNonNumericSubstring(s, s.indexOf(' ') + 1, ':');
- if (s.indexOf(':') != -1) {
- // clean second colon
- s = replaceFirstNonNumericSubstring(s, s.indexOf(':') + 1, ':');
- }
- }
-
- final int maxMonth = 12;
- final int maxDay = 31;
-
- // Parse the date
- int firstDash = s.indexOf('-');
- int secondDash = s.indexOf('-', firstDash + 1);
- int dividingSpace = s.indexOf(' ');
-
- // Parse the time
- int firstColon = s.indexOf(':', dividingSpace + 1);
- int secondColon = s.indexOf(':', firstColon + 1);
- int period = s.indexOf('.', secondColon + 1);
-
- int year = 0;
- int month = 0;
- int day = 0;
- int hour = 0;
- int minute = 0;
- int second = 0;
-
- // Get the date
- int len = s.length();
- boolean parsedDate = false;
- if (firstDash > 0 && secondDash > firstDash) {
- year = Integer.parseInt(s.substring(0, firstDash));
- month = Integer.parseInt(s.substring(firstDash + 1, secondDash));
- if (dividingSpace != -1) {
- day = Integer.parseInt(s.substring(secondDash + 1, dividingSpace));
- } else {
- day = Integer.parseInt(s.substring(secondDash + 1, len));
- }
-
- if ((month >= 1 && month <= maxMonth) && (day >= 1 && day <= maxDay)) {
- parsedDate = true;
- }
- }
- if (!parsedDate) {
- throw new java.lang.IllegalArgumentException("Cannot parse the date from " + s);
- }
-
- // Get the time. Hour, minute, second and colons are all optional
- if (dividingSpace != -1 && dividingSpace < len - 1) {
- if (firstColon == -1) {
- hour = Integer.parseInt(s.substring(dividingSpace + 1, len));
- } else {
- hour = Integer.parseInt(s.substring(dividingSpace + 1, firstColon));
- if (firstColon < len - 1) {
- if (secondColon == -1) {
- minute = Integer.parseInt(s.substring(firstColon + 1, len));
- } else {
- minute = Integer.parseInt(s.substring(firstColon + 1, secondColon));
- if (secondColon < len - 1) {
- if (period == -1) {
- second = Integer.parseInt(s.substring(secondColon + 1, len));
- } else {
- second = Integer.parseInt(s.substring(secondColon + 1, period));
- }
- }
- }
- }
- }
- }
-
- StringBuilder cleanedTimestamp = new StringBuilder();
- cleanedTimestamp =
- cleanedTimestamp.append(
- String.format(
- "%04d-%02d-%02d %02d:%02d:%02d",
- year, month, day, hour, minute, second));
-
- if (period != -1 && period < len - 1) {
- cleanedTimestamp = cleanedTimestamp.append(".").append(s.substring(period + 1));
- }
-
- return cleanedTimestamp.toString();
- }
-
- /**
- * Replace the first non-numeric substring.
- *
- * @param s the original string
- * @param startIndex the beginning index, inclusive
- * @param c the new character
- * @return
- */
- private String replaceFirstNonNumericSubstring(String s, int startIndex, char c) {
- StringBuilder sb = new StringBuilder();
- sb.append(s.substring(0, startIndex));
-
- String rest = s.substring(startIndex);
- sb.append(rest.replaceFirst("[^\\d]+", Character.toString(c)));
- return sb.toString();
- }
-
- public ColumnEditor setColumnDefaultValue(ColumnEditor columnEditor) {
- final Column column = columnEditor.create();
-
- // if converters is not null and the default value is not null, we need to convert default
- // value
- if (converters != null && columnEditor.defaultValue() != null) {
- Object defaultValue = columnEditor.defaultValue();
- final SchemaBuilder schemaBuilder = converters.schemaBuilder(column);
- if (schemaBuilder == null) {
- return columnEditor;
- }
- final Schema schema = schemaBuilder.build();
- // In order to get the valueConverter for this column, we have to create a field;
- // The index value -1 in the field will never used when converting default value;
- // So we can set any number here;
- final Field field = new Field(columnEditor.name(), -1, schema);
- final ValueConverter valueConverter =
- converters.converter(columnEditor.create(), field);
- if (defaultValue instanceof String) {
- defaultValue = convert(column, (String) defaultValue);
- }
- defaultValue = valueConverter.convert(defaultValue);
- columnEditor.defaultValue(defaultValue);
- }
- return columnEditor;
- }
-}
diff --git a/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java b/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
index f7ac0bded8e..70d22049006 100644
--- a/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
+++ b/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
@@ -56,9 +56,9 @@
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
import javax.net.ssl.X509TrustManager;
-import java.io.FileInputStream;
import java.io.IOException;
import java.security.GeneralSecurityException;
import java.security.KeyStore;
@@ -83,17 +83,9 @@
import static io.debezium.util.Strings.isNullOrEmpty;
-/**
- * Copied from Debezium project to fix https://github.com/ververica/flink-cdc-connectors/issues/939.
- *
- * Line 268 ~ 270: Clean cache on rotate event to prevent it from growing indefinitely. We should
- * remove this class after we bumped a higher debezium version where the
- * https://issues.redhat.com/browse/DBZ-5126 has been fixed.
- *
- * Line 1386 : Add more error details for some exceptions.
- */
+/** Copied from Debezium project to add more error details for some exceptions in line 1465. */
public class MySqlStreamingChangeEventSource
- implements StreamingChangeEventSource Line 66~77: use the actual default string value when the sql contains COLLATE. We should
+ * Line 48~59: use the actual default string value when the sql contains COLLATE. We should
* remove this class after we bumped a higher debezium version where the
* https://issues.redhat.com/browse/DBZ-5587 has been fixed.
*/
@@ -28,32 +25,17 @@ public class DefaultValueParserListener extends MySqlParserBaseListener {
private final ColumnEditor columnEditor;
private final AtomicReference This file is override to fix logger mining session stopped due to 'No more data to read from
- * socket' exception. please see more discussion under
- * https://github.com/debezium/debezium/pull/3118, We should remove this class since we bumped
- * higher debezium version after 1.8.1.Final where the issue has been fixed.
- */
-public class OracleErrorHandler extends ErrorHandler {
-
- private static final List Line 356: Replace < condition with <= to be able to catch ongoing transactions during snapshot
+ * if current SCN points to START/INSERT/DELETE/UPDATE event.
+ */
+public class LogMinerAdapter extends AbstractStreamingAdapter {
+
+ private static final Duration GET_TRANSACTION_SCN_PAUSE = Duration.ofSeconds(1);
+
+ private static final int GET_TRANSACTION_SCN_ATTEMPTS = 5;
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(LogMinerAdapter.class);
+
+ public static final String TYPE = "logminer";
+
+ public LogMinerAdapter(OracleConnectorConfig connectorConfig) {
+ super(connectorConfig);
+ }
+
+ @Override
+ public String getType() {
+ return TYPE;
+ }
+
+ @Override
+ public HistoryRecordComparator getHistoryRecordComparator() {
+ return new HistoryRecordComparator() {
+ @Override
+ protected boolean isPositionAtOrBefore(Document recorded, Document desired) {
+ return resolveScn(recorded).compareTo(resolveScn(desired)) < 1;
+ }
+ };
+ }
+
+ @Override
+ public OffsetContext.Loader Emits change record based on a single {@link LogMinerDmlEntry} event.
- *
- * This class overrides the emit methods to put the ROWID in the header.
- *
- * Line 59 ~ 257: add ROWID and emit methods.
- */
+/** Emits change records based on an event read from Oracle LogMiner. */
public class LogMinerChangeRecordEmitter extends BaseChangeRecordEmitter