This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new ce15df9b6 [ISSUE #4439]Support source connector jdbc handle CDC for
mysql (#4457)
ce15df9b6 is described below
commit ce15df9b6309a7a9f6bb2c43a44faf317be81e69
Author: mxsm <[email protected]>
AuthorDate: Tue Sep 26 14:40:37 2023 +0800
[ISSUE #4439]Support source connector jdbc handle CDC for mysql (#4457)
* [ISSUE #4439]Support source connector jdbc handle CDC for mysql
* fix code style
---
.../apache/eventmesh/common/utils/LogUtils.java | 214 ++++++
.../eventmesh/connector/jdbc/DataChanges.java | 71 ++
.../eventmesh/connector/jdbc/JdbcConnectData.java | 29 +
.../apache/eventmesh/connector/jdbc/Payload.java | 21 +-
.../apache/eventmesh/connector/jdbc/Schema.java | 17 +-
.../connector/jdbc/event/DeleteDataEvent.java | 2 +-
.../jdbc/event/GeneralDataChangeEvent.java | 3 +-
.../jdbc/event/GeneralSchemaChangeEvent.java | 2 +-
.../connector/jdbc/event/UpdateDataEvent.java | 2 +-
.../connector/jdbc/source/JdbcSourceConnector.java | 21 +-
.../jdbc/source/SourceEventMeshJdbcEventTask.java | 6 +-
.../listener/ColumnDefinitionParserListener.java | 3 +-
.../mysql/listener/CreateTableParserListener.java | 2 +
.../mysql/listener/TableBaseParserListener.java | 10 +-
.../jdbc/source/dialect/cdc/AbstractCdcEngine.java | 134 ++++
.../dialect/cdc/AbstractCdcEngineFactory.java} | 14 +-
.../source/dialect/cdc/mysql/MysqlCdcEngine.java | 809 +++++++++++++++++++++
.../dialect/cdc/mysql/MysqlCdcEngineFactory.java | 54 ++
.../source/dialect/cdc/mysql/RowDeserializers.java | 452 ++++++++++++
.../mysql/{GtidSet.java => EventMeshGtidSet.java} | 34 +-
.../dialect/mysql/MysqlConstants.java} | 27 +-
.../source/dialect/mysql/MysqlJdbcContext.java | 28 +-
.../jdbc/source/dialect/mysql/MysqlSourceInfo.java | 9 +-
.../dialect/snapshot/AbstractSnapshotEngine.java | 36 +-
.../snapshot/mysql/MysqlSnapshotEngine.java | 15 +-
.../table/catalog/AbstractColumnEditorImpl.java | 19 +
.../connector/jdbc/table/catalog/ColumnEditor.java | 8 +
.../connector/jdbc/table/catalog/TableId.java | 4 +
.../connector/jdbc/table/catalog/TableSchema.java | 23 +-
.../table/catalog/mysql/MysqlColumnEditorImpl.java | 2 +-
.../jdbc/table/catalog/mysql/MysqlTableSchema.java | 7 +-
...nector.jdbc.source.dialect.cdc.CdcEngineFactory | 2 +-
32 files changed, 1966 insertions(+), 114 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/LogUtils.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/LogUtils.java
new file mode 100644
index 000000000..227cdabbd
--- /dev/null
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/LogUtils.java
@@ -0,0 +1,214 @@
+/*
+ * 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.eventmesh.common.utils;
+
+import org.slf4j.Logger;
+import org.slf4j.Marker;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public final class LogUtils {
+
+ public static boolean isTraceEnabled(Logger logger) {
+ return logger.isTraceEnabled();
+ }
+
+ public static boolean isTraceEnabled(Logger logger, Marker marker) {
+ return logger.isTraceEnabled(marker);
+ }
+
+ public static void trace(Logger logger, String msg) {
+ if (isTraceEnabled(logger)) {
+ logger.trace(msg);
+ }
+ }
+
+ public static void trace(Logger logger, String format, Object arg) {
+ if (isTraceEnabled(logger)) {
+ logger.trace(format, arg);
+ }
+ }
+
+ public static void trace(Logger logger, String msg, Throwable t) {
+ if (isTraceEnabled(logger)) {
+ logger.trace(msg, t);
+ }
+ }
+
+ public static void trace(Logger logger, String format, Object...
arguments) {
+ if (isTraceEnabled(logger)) {
+ logger.trace(format, arguments);
+ }
+ }
+
+ public static void trace(Logger logger, String format, Object arg1, Object
arg2) {
+ if (isTraceEnabled(logger)) {
+ logger.trace(format, arg1, arg2);
+ }
+ }
+
+ public static void trace(Logger logger, Marker marker, String msg) {
+ if (isTraceEnabled(logger, marker)) {
+ logger.trace(marker, msg);
+ }
+ }
+
+ public static void trace(Logger logger, Marker marker, String format,
Object arg) {
+ if (isTraceEnabled(logger, marker)) {
+ logger.trace(marker, format, arg);
+ }
+ }
+
+ public static boolean isDebugEnabled(Logger logger) {
+ return logger.isDebugEnabled();
+ }
+
+ public static void debug(Logger logger, String msg) {
+ if (isDebugEnabled(logger)) {
+ logger.debug(msg);
+ }
+ }
+
+ public static void debug(Logger logger, String format, Object arg) {
+ if (isDebugEnabled(logger)) {
+ logger.debug(format, arg);
+ }
+ }
+
+ public static void debug(Logger logger, String format, Object arg1, Object
arg2) {
+ if (isDebugEnabled(logger)) {
+ logger.debug(format, arg1, arg2);
+ }
+ }
+
+ public static void debug(Logger logger, String format, Object...
arguments) {
+ if (isDebugEnabled(logger)) {
+ logger.debug(format, arguments);
+ }
+ }
+
+ public static void debug(Logger logger, String msg, Throwable t) {
+ if (isDebugEnabled(logger)) {
+ logger.debug(msg, t);
+ }
+ }
+
+ public static boolean isInfoEnabled(Logger logger) {
+ return logger.isInfoEnabled();
+ }
+
+ public static void info(Logger logger, String msg) {
+ if (isInfoEnabled(logger)) {
+ logger.info(msg);
+ }
+ }
+
+ public static void info(Logger logger, String format, Object arg) {
+ if (isInfoEnabled(logger)) {
+ logger.info(format, arg);
+ }
+ }
+
+ public static void info(Logger logger, String format, Object arg1, Object
arg2) {
+ if (isInfoEnabled(logger)) {
+ logger.info(format, arg1, arg2);
+ }
+ }
+
+ public static void info(Logger logger, String format, Object... arguments)
{
+ if (isInfoEnabled(logger)) {
+ logger.info(format, arguments);
+ }
+ }
+
+ public static void info(Logger logger, String msg, Throwable t) {
+ if (isInfoEnabled(logger)) {
+ logger.info(msg, t);
+ }
+ }
+
+ public static boolean isWarnEnabled(Logger logger) {
+ return logger.isWarnEnabled();
+ }
+
+ public static void warn(Logger logger, String msg) {
+ if (isWarnEnabled(logger)) {
+ logger.warn(msg);
+ }
+ }
+
+ public static void warn(Logger logger, String format, Object arg) {
+ if (isWarnEnabled(logger)) {
+ logger.warn(format, arg);
+ }
+ }
+
+ public static void warn(Logger logger, String format, Object arg1, Object
arg2) {
+ if (isWarnEnabled(logger)) {
+ logger.warn(format, arg1, arg2);
+ }
+ }
+
+ public static void warn(Logger logger, String format, Object... arguments)
{
+ if (isWarnEnabled(logger)) {
+ logger.warn(format, arguments);
+ }
+ }
+
+ public static void warn(Logger logger, String msg, Throwable t) {
+ if (isWarnEnabled(logger)) {
+ logger.warn(msg, t);
+ }
+ }
+
+ public static boolean isErrorEnabled(Logger logger) {
+ return logger.isErrorEnabled();
+ }
+
+ public static void error(Logger logger, String msg) {
+ if (isErrorEnabled(logger)) {
+ logger.error(msg);
+ }
+ }
+
+ public static void error(Logger logger, String format, Object arg) {
+ if (isErrorEnabled(logger)) {
+ logger.error(format, arg);
+ }
+ }
+
+ public static void error(Logger logger, String format, Object arg1, Object
arg2) {
+ if (isErrorEnabled(logger)) {
+ logger.error(format, arg1, arg2);
+ }
+ }
+
+ public static void error(Logger logger, String format, Object...
arguments) {
+ if (isErrorEnabled(logger)) {
+ logger.error(format, arguments);
+ }
+ }
+
+ public static void error(Logger logger, String msg, Throwable t) {
+ if (isErrorEnabled(logger)) {
+ logger.error(msg, t);
+ }
+ }
+
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DataChanges.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DataChanges.java
new file mode 100644
index 000000000..cb0298e91
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DataChanges.java
@@ -0,0 +1,71 @@
+/*
+ * 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.eventmesh.connector.jdbc;
+
+import lombok.Data;
+
+@Data
+public class DataChanges {
+
+ private Object after;
+
+ private Object before;
+
+ private String type;
+
+ public DataChanges(Object after, Object before) {
+ this.after = after;
+ this.before = before;
+ }
+
+ public DataChanges(Object after, Object before, String type) {
+ this.after = after;
+ this.before = before;
+ this.type = type;
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ public static class Builder {
+
+ private String type;
+ private Object after;
+ private Object before;
+
+ public Builder withType(String type) {
+ this.type = type;
+ return this;
+ }
+
+ public Builder withAfter(Object after) {
+ this.after = after;
+ return this;
+ }
+
+ public Builder withBefore(Object before) {
+ this.before = before;
+ return this;
+ }
+
+ public DataChanges build() {
+ return new DataChanges(after, before, type);
+ }
+ }
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
index f70819212..3dca4f42f 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
@@ -19,10 +19,23 @@ package org.apache.eventmesh.connector.jdbc;
public final class JdbcConnectData {
+ public static final byte DATA_CHANGES = 1;
+
+ public static final byte SCHEMA_CHANGES = 1 << 1;
+
private Payload payload = new Payload();
private Schema schema;
+ private byte type;
+
+ public JdbcConnectData() {
+ }
+
+ public JdbcConnectData(byte type) {
+ this.type = type;
+ }
+
public Payload getPayload() {
return payload;
}
@@ -38,4 +51,20 @@ public final class JdbcConnectData {
public void setSchema(Schema schema) {
this.schema = schema;
}
+
+ public byte getType() {
+ return type;
+ }
+
+ public void setType(byte type) {
+ this.type = type;
+ }
+
+ public void markDataChanges() {
+ this.type |= DATA_CHANGES;
+ }
+
+ public void markSchemaChanges() {
+ this.type |= SCHEMA_CHANGES;
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Payload.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Payload.java
index 262648c5e..47a5b42fe 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Payload.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Payload.java
@@ -23,6 +23,15 @@ import java.util.HashMap;
public final class Payload extends HashMap<String, Object> {
+ public static final String AFTER_FIELD = "after";
+
+ public static final String BEFORE_FIELD = "before";
+
+ public static final String SOURCE = "source";
+
+ public static final String DDL = "ddl";
+
+
/**
* Constructs an empty <code>HashMap</code> with the default initial
capacity (16) and the default load factor (0.75).
*/
@@ -31,12 +40,12 @@ public final class Payload extends HashMap<String, Object> {
}
public Payload withSource(SourceMateData source) {
- this.put("source", source);
+ this.put(SOURCE, source);
return this;
}
public Payload withDdl(String ddl) {
- this.put("ddl", ddl);
+ this.put(DDL, ddl);
return this;
}
@@ -45,13 +54,13 @@ public final class Payload extends HashMap<String, Object> {
return this;
}
- public Payload withAfterValue(Object value) {
- this.put("after", value);
+ public Payload withDataChanges(DataChanges dataChanges) {
+ this.put("dataChanges", dataChanges);
return this;
}
public SourceMateData ofSourceMateData() {
- return (SourceMateData) super.get("source");
+ return (SourceMateData) super.get(SOURCE);
}
public static Builder builder() {
@@ -72,7 +81,7 @@ public final class Payload extends HashMap<String, Object> {
}
public Builder withSource(SourceMateData source) {
- payload.put("source", source);
+ payload.put(SOURCE, source);
return this;
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
index 284bc14b1..084f9759b 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
@@ -17,16 +17,25 @@
package org.apache.eventmesh.connector.jdbc;
+import java.util.ArrayList;
import java.util.List;
-import lombok.AllArgsConstructor;
import lombok.Data;
-import lombok.NoArgsConstructor;
@Data
-@AllArgsConstructor
-@NoArgsConstructor
public class Schema {
private List<Field> fields;
+
+ public Schema(List<Field> fields) {
+ this.fields = fields;
+ }
+
+ public Schema() {
+ this.fields = new ArrayList<>();
+ }
+
+ public void add(Field field) {
+ this.fields.add(field);
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DeleteDataEvent.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DeleteDataEvent.java
index 68f23ece8..9e0d38e49 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DeleteDataEvent.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DeleteDataEvent.java
@@ -33,6 +33,6 @@ public class DeleteDataEvent extends GeneralDataChangeEvent {
@Override
public DataChangeEventType getDataChangeEventType() {
- return DataChangeEventType.INSERT;
+ return DataChangeEventType.DELETE;
}
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/GeneralDataChangeEvent.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/GeneralDataChangeEvent.java
index f7bcf24d3..c6a922237 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/GeneralDataChangeEvent.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/GeneralDataChangeEvent.java
@@ -22,9 +22,8 @@ import
org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
public abstract class GeneralDataChangeEvent extends AbstractEvent implements
DataChangeEvent {
-
public GeneralDataChangeEvent(TableId tableId) {
- super(tableId);
+ super(tableId, new JdbcConnectData(JdbcConnectData.DATA_CHANGES));
}
public GeneralDataChangeEvent(TableId tableId, JdbcConnectData data) {
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/GeneralSchemaChangeEvent.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/GeneralSchemaChangeEvent.java
index f07fee84b..80996203c 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/GeneralSchemaChangeEvent.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/GeneralSchemaChangeEvent.java
@@ -23,7 +23,7 @@ import
org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
public abstract class GeneralSchemaChangeEvent extends AbstractEvent
implements SchemaChangeEvent {
public GeneralSchemaChangeEvent(TableId tableId) {
- super(tableId);
+ super(tableId, new JdbcConnectData(JdbcConnectData.SCHEMA_CHANGES));
}
public GeneralSchemaChangeEvent(TableId tableId, JdbcConnectData data) {
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/UpdateDataEvent.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/UpdateDataEvent.java
index dc1ea9c45..d6c50859b 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/UpdateDataEvent.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/UpdateDataEvent.java
@@ -36,7 +36,7 @@ public class UpdateDataEvent extends GeneralDataChangeEvent {
@Override
public DataChangeEventType getDataChangeEventType() {
- return DataChangeEventType.INSERT;
+ return DataChangeEventType.UPDATE;
}
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnector.java
index 618f56339..bc3b55006 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnector.java
@@ -22,6 +22,7 @@ import org.apache.eventmesh.connector.jdbc.event.Event;
import org.apache.eventmesh.connector.jdbc.source.config.JdbcSourceConfig;
import
org.apache.eventmesh.connector.jdbc.source.dialect.DatabaseDialectFactory;
import org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngine;
+import org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngineFactory;
import
org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.SnapshotEngine;
import
org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.SnapshotEngineFactory;
import
org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.SnapshotResult;
@@ -34,6 +35,8 @@ import
org.apache.eventmesh.openconnect.api.connector.SourceConnector;
import org.apache.eventmesh.openconnect.api.connector.SourceConnectorContext;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.commons.collections4.CollectionUtils;
+
import java.util.List;
import java.util.Set;
@@ -117,6 +120,20 @@ public class JdbcSourceConnector extends SourceConnector {
this.snapshotEngine.registerSnapshotEventConsumer(this::eventConsumer);
this.snapshotEngine.init();
+ // Get the CDC engine factory and create the CDC engine.
+ final CdcEngineFactory cdcEngineFactory =
JdbcAllFactoryLoader.getCdcEngineFactory(databaseType);
+ // Check if the CDC engine factory supports the JDBC protocol.
+ if
(!cdcEngineFactory.acceptJdbcProtocol(this.databaseDialect.jdbcProtocol())) {
+ throw new IllegalArgumentException("CdcEngineFactory not supports
" + databaseType);
+ }
+ // Set the CDC engine and register the CDC event consumer.
+ this.cdcEngine = cdcEngineFactory.createCdcEngine(this.sourceConfig,
this.databaseDialect);
+ if (CollectionUtils.isEmpty(this.cdcEngine.getHandledTables())) {
+ throw new RuntimeException("No database tables need to be
processed");
+ }
+ this.cdcEngine.registerCdcEventConsumer(this::eventConsumer);
+ this.cdcEngine.init();
+
Set<TableId> handledTables = this.snapshotEngine.getHandledTables();
// Create the task manager and dispatcher.
@@ -150,8 +167,8 @@ public class JdbcSourceConnector extends SourceConnector {
//success and skip status can run cdc engine
if (result.getStatus() != SnapshotResultStatus.ABORTED) {
log.info("Start Cdc Engine to handle cdc event");
- /*this.cdcEngine.setContext(result.getContext());
- this.cdcEngine.start();*/
+ this.cdcEngine.setContext(result.getContext());
+ this.cdcEngine.start();
}
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceEventMeshJdbcEventTask.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceEventMeshJdbcEventTask.java
index 7c2b0d717..f24cccc65 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceEventMeshJdbcEventTask.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceEventMeshJdbcEventTask.java
@@ -53,11 +53,11 @@ public class SourceEventMeshJdbcEventTask extends
AbstractEventMeshJdbcEventTask
public void run() {
while (isRunning) {
try {
- Event snapshotEvent = eventBlockingQueue.poll(5,
TimeUnit.SECONDS);
- if (Objects.isNull(snapshotEvent)) {
+ Event event = eventBlockingQueue.poll(5, TimeUnit.SECONDS);
+ if (Objects.isNull(event)) {
continue;
}
- eventHandler.handle(snapshotEvent);
+ eventHandler.handle(event);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/ColumnDefinitionParserListener.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/ColumnDefinitionParserListener.java
index 499ac0910..4877360f9 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/ColumnDefinitionParserListener.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/ColumnDefinitionParserListener.java
@@ -130,8 +130,7 @@ public class ColumnDefinitionParserListener extends
MySqlParserBaseListener {
@SuppressWarnings("unchecked")
public void exitColumnDefinition(ColumnDefinitionContext ctx) {
if (!ignoreColumn.get().booleanValue()) {
- this.tableEditor.addColumns(columnEditor.build());
- ignoreColumn.set(false);
+ this.ignoreColumn.set(false);
}
//When exit column definition needs to remove
DefaultValueParserListener from listener list
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/CreateTableParserListener.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/CreateTableParserListener.java
index 2ed93cfba..cf117952b 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/CreateTableParserListener.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/CreateTableParserListener.java
@@ -120,6 +120,8 @@ public class CreateTableParserListener extends
TableBaseParserListener {
payload.withSource(sourceMateData).withDdl(ddl).withCatalogChanges(changes);
parser.handleEvent(event);
}, tableEditor);
+ //reset column order
+ columnOrder.set(1);
super.exitColumnCreateTable(ctx);
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/TableBaseParserListener.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/TableBaseParserListener.java
index b978214b1..98d35fa6a 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/TableBaseParserListener.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/antlr4/mysql/listener/TableBaseParserListener.java
@@ -35,6 +35,7 @@ import org.apache.commons.collections4.CollectionUtils;
import java.util.List;
import java.util.Objects;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.antlr.v4.runtime.tree.ParseTreeListener;
@@ -49,6 +50,8 @@ public class TableBaseParserListener extends
MySqlParserBaseListener {
protected ColumnDefinitionParserListener columnDefinitionListener;
+ protected AtomicInteger columnOrder = new AtomicInteger(1);
+
public TableBaseParserListener(List<ParseTreeListener> listeners,
MysqlAntlr4DdlParser parser) {
this.listeners = listeners;
this.parser = parser;
@@ -95,8 +98,11 @@ public class TableBaseParserListener extends
MySqlParserBaseListener {
@Override
public void exitColumnDeclaration(ColumnDeclarationContext ctx) {
- parser.runIfAllNotNull(() ->
tableEditor.addColumns(columnDefinitionListener.getColumnEditor().build()),
tableEditor,
- columnDefinitionListener);
+ parser.runIfAllNotNull(() -> {
+ MysqlColumnEditor columnEditor =
columnDefinitionListener.getColumnEditor();
+ columnEditor.withOrder(columnOrder.getAndIncrement());
+ tableEditor.addColumns(columnEditor.build());
+ }, tableEditor, columnDefinitionListener);
super.exitColumnDeclaration(ctx);
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/AbstractCdcEngine.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/AbstractCdcEngine.java
new file mode 100644
index 000000000..c164cb7ee
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/AbstractCdcEngine.java
@@ -0,0 +1,134 @@
+/*
+ * 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.eventmesh.connector.jdbc.source.dialect.cdc;
+
+
+import org.apache.eventmesh.common.ThreadWrapper;
+import org.apache.eventmesh.connector.jdbc.DatabaseDialect;
+import org.apache.eventmesh.connector.jdbc.JdbcContext;
+import org.apache.eventmesh.connector.jdbc.ddl.DdlParser;
+import org.apache.eventmesh.connector.jdbc.source.config.JdbcSourceConfig;
+import org.apache.eventmesh.connector.jdbc.source.config.SourceConnectorConfig;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
+
+import org.apache.commons.collections4.CollectionUtils;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public abstract class AbstractCdcEngine<Parse extends DdlParser, Ctx extends
JdbcContext, DbDialect extends DatabaseDialect> extends
+ ThreadWrapper implements CdcEngine<Ctx> {
+
+ protected final JdbcSourceConfig jdbcSourceConfig;
+
+ protected final DbDialect databaseDialect;
+
+ protected final SourceConnectorConfig sourceConnectorConfig;
+
+ private final Set<TableId> includeDatabaseTable = new HashSet<>(64);
+
+ public AbstractCdcEngine(SourceConfig config, DbDialect databaseDialect) {
+ if (!(config instanceof JdbcSourceConfig)) {
+ throw new IllegalArgumentException("config ");
+ }
+ this.jdbcSourceConfig = (JdbcSourceConfig) config;
+ this.databaseDialect = databaseDialect;
+ this.sourceConnectorConfig =
this.jdbcSourceConfig.getSourceConnectorConfig();
+
+ calculateNeedHandleTable();
+ }
+
+ @Override
+ public Set<TableId> getHandledTables() {
+ return includeDatabaseTable;
+ }
+
+
+ protected Set<TableId> calculateNeedHandleTable() {
+ // Get the database and table include and exclude lists from the
connector configuration
+ List<String> databaseIncludeList =
sourceConnectorConfig.getDatabaseIncludeList();
+
+ // If the database include list is empty, get a list of all databases
and use that as the include list
+ if (CollectionUtils.isEmpty(databaseIncludeList)) {
+ List<String> allDatabases = databaseDialect.listDatabases();
+ databaseIncludeList = new ArrayList<>(allDatabases);
+ }
+ Set<String> defaultExcludeDatabase = defaultExcludeDatabase();
+ if (CollectionUtils.isNotEmpty(defaultExcludeDatabase)) {
+ databaseIncludeList.removeAll(defaultExcludeDatabase);
+ }
+
+ List<String> databaseExcludeList =
sourceConnectorConfig.getDatabaseExcludeList();
+ // Remove the default excluded databases from the include list
+ if (CollectionUtils.isNotEmpty(databaseExcludeList)) {
+ databaseIncludeList.removeAll(databaseExcludeList);
+ }
+
+ List<String> tableIncludeList =
sourceConnectorConfig.getTableIncludeList();
+ // Create a list of included tables based on the table include list
+ List<TableId> includeTableList = new ArrayList<>();
+ if (CollectionUtils.isNotEmpty(tableIncludeList)) {
+ List<TableId> tableIdList = buildTableId(tableIncludeList);
+ includeTableList.addAll(tableIdList);
+ }
+
+ // If the table include list is empty, get a list of all tables for
each database in the include list
+ if (CollectionUtils.isEmpty(tableIncludeList)) {
+ for (String database : databaseIncludeList) {
+ try {
+ List<TableId> tableIds =
databaseDialect.listTables(database);
+ includeTableList.addAll(tableIds);
+ } catch (SQLException e) {
+ log.warn("List database[{}] table error", database, e);
+ }
+ }
+ }
+
+ List<String> tableExcludeList =
sourceConnectorConfig.getTableExcludeList();
+ // Remove any tables in the exclude list from the included tables list
+ if (CollectionUtils.isNotEmpty(tableExcludeList)) {
+ includeTableList.removeAll(buildTableId(tableExcludeList));
+ }
+
+ includeDatabaseTable.addAll(includeTableList);
+
+ return includeDatabaseTable;
+ }
+
+
+ private List<TableId> buildTableId(List<String> tables) {
+ return Optional.ofNullable(tables).orElse(new
ArrayList<>(0)).stream().map(table -> {
+ String[] split = table.split("\\.");
+ return new TableId(split[0], null, split[1]);
+ }).collect(Collectors.toList());
+ }
+
+ protected abstract Set<String> defaultExcludeDatabase();
+
+ protected abstract Parse getDdlParser();
+}
+
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/AbstractCdcEngineFactory.java
similarity index 75%
copy from
eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
copy to
eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/AbstractCdcEngineFactory.java
index 284bc14b1..d00d4c56f 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/AbstractCdcEngineFactory.java
@@ -15,18 +15,8 @@
* limitations under the License.
*/
-package org.apache.eventmesh.connector.jdbc;
+package org.apache.eventmesh.connector.jdbc.source.dialect.cdc;
-import java.util.List;
+public abstract class AbstractCdcEngineFactory implements CdcEngineFactory {
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-public class Schema {
-
- private List<Field> fields;
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngine.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngine.java
new file mode 100644
index 000000000..8a29bf420
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngine.java
@@ -0,0 +1,809 @@
+/*
+ * 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.eventmesh.connector.jdbc.source.dialect.cdc.mysql;
+
+import org.apache.eventmesh.common.EventMeshThreadFactory;
+import org.apache.eventmesh.common.utils.LogUtils;
+import org.apache.eventmesh.connector.jdbc.DataChanges;
+import org.apache.eventmesh.connector.jdbc.DataChanges.Builder;
+import org.apache.eventmesh.connector.jdbc.Field;
+import org.apache.eventmesh.connector.jdbc.Payload;
+import org.apache.eventmesh.connector.jdbc.Schema;
+import org.apache.eventmesh.connector.jdbc.config.JdbcConfig;
+import
org.apache.eventmesh.connector.jdbc.connection.mysql.MysqlJdbcConnection;
+import org.apache.eventmesh.connector.jdbc.event.DeleteDataEvent;
+import org.apache.eventmesh.connector.jdbc.event.EventConsumer;
+import org.apache.eventmesh.connector.jdbc.event.GeneralDataChangeEvent;
+import org.apache.eventmesh.connector.jdbc.event.InsertDataEvent;
+import org.apache.eventmesh.connector.jdbc.event.UpdateDataEvent;
+import org.apache.eventmesh.connector.jdbc.source.config.JdbcSourceConfig;
+import org.apache.eventmesh.connector.jdbc.source.config.MysqlConfig;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.antlr4.mysql.MysqlAntlr4DdlParser;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.cdc.AbstractCdcEngine;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.cdc.mysql.RowDeserializers.DeleteRowsEventMeshDeserializer;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.cdc.mysql.RowDeserializers.UpdateRowsEventMeshDeserializer;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.cdc.mysql.RowDeserializers.WriteRowsEventMeshDeserializer;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.EventDataDeserializationExceptionData;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.EventMeshGtidSet;
+import org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlConstants;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlDatabaseDialect;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlJdbcContext;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlSourceMateData;
+import org.apache.eventmesh.connector.jdbc.table.catalog.Column;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableSchema;
+import org.apache.eventmesh.connector.jdbc.table.type.Pair;
+import org.apache.eventmesh.openconnect.api.config.Config;
+
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import com.github.shyiko.mysql.binlog.BinaryLogClient;
+import com.github.shyiko.mysql.binlog.BinaryLogClient.LifecycleListener;
+import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
+import com.github.shyiko.mysql.binlog.event.Event;
+import com.github.shyiko.mysql.binlog.event.EventData;
+import com.github.shyiko.mysql.binlog.event.EventHeader;
+import com.github.shyiko.mysql.binlog.event.EventHeaderV4;
+import com.github.shyiko.mysql.binlog.event.EventType;
+import com.github.shyiko.mysql.binlog.event.GtidEventData;
+import com.github.shyiko.mysql.binlog.event.QueryEventData;
+import com.github.shyiko.mysql.binlog.event.RotateEventData;
+import com.github.shyiko.mysql.binlog.event.TableMapEventData;
+import com.github.shyiko.mysql.binlog.event.TransactionPayloadEventData;
+import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
+import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
+import com.github.shyiko.mysql.binlog.event.XidEventData;
+import
com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializationException;
+import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
+import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class MysqlCdcEngine extends AbstractCdcEngine<MysqlAntlr4DdlParser,
MysqlJdbcContext, MysqlDatabaseDialect> {
+
+ private BinaryLogClient client;
+
+ private BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<>(10000);
+
+ private final EnumMap<EventType, Consumer<Event>> eventHandlers = new
EnumMap<>(EventType.class);
+
+ private Map<Long/*table number*/, TableId> tableIdMap = new HashMap<>(64);
+
+ private MysqlJdbcContext context;
+
+ private List<EventConsumer> consumers = new ArrayList<>(16);
+
+ private MysqlAntlr4DdlParser ddlParser;
+
+ private MysqlJdbcConnection connection;
+
+ private com.github.shyiko.mysql.binlog.GtidSet localGtidSet;
+
+ public MysqlCdcEngine(Config config, MysqlDatabaseDialect databaseDialect)
{
+ super((JdbcSourceConfig) config, databaseDialect);
+ this.ddlParser = new MysqlAntlr4DdlParser(false, false,
getHandledTables(), (JdbcSourceConfig) config);
+ this.connection = databaseDialect.getConnection();
+ }
+
+ /**
+ * Initializes the CDC Engine.
+ */
+ @Override
+ public void init() {
+ final JdbcConfig jdbcConfig =
this.sourceConnectorConfig.getJdbcConfig();
+ client = new BinaryLogClient(jdbcConfig.getHostname(),
jdbcConfig.getPort(), jdbcConfig.getUser(), jdbcConfig.getPassword());
+ client.setThreadFactory(new
EventMeshThreadFactory("mysql-binlog-client"));
+ final MysqlConfig mysqlConfig =
this.sourceConnectorConfig.getMysqlConfig();
+ client.setServerId(mysqlConfig.getServerId());
+ client.setKeepAlive(mysqlConfig.isKeepAlive());
+ final long keepAliveInterval = mysqlConfig.getKeepAliveInterval();
+ client.setKeepAliveInterval(keepAliveInterval);
+
+ final Map<Long, TableMapEventData> tableMapEventByTableId = new
HashMap<>(32);
+
+ // mysql dev url:https://dev.mysql.com/doc/dev/mysql-server/latest/
+ EventDeserializer eventDeserializer = new EventDeserializer() {
+ /**
+ * @param inputStream input stream to fetch event from
+ * @return deserialized event or null in case of end-of-stream
+ * @throws IOException if connection gets closed
+ */
+ @Override
+ public Event nextEvent(ByteArrayInputStream inputStream) throws
IOException {
+ try {
+ // Delegate to the superclass
+ Event event = super.nextEvent(inputStream);
+ if (log.isDebugEnabled()) {
+ log.debug("MYSQL Binlog---EventType={}, EventData={}",
event.getHeader().getEventType(), event);
+ }
+ // We have to record the most recent TableMapEventData for
each table number for our custom deserializers
+ if (event.getHeader().getEventType() ==
EventType.TABLE_MAP) {
+ TableMapEventData tableMapEvent = event.getData();
+ tableMapEventByTableId.put(tableMapEvent.getTableId(),
tableMapEvent);
+ }
+
+ // Handle for transaction payload and capture the table
map event and add it to the map
+ if (event.getHeader().getEventType() ==
EventType.TRANSACTION_PAYLOAD) {
+ TransactionPayloadEventData
transactionPayloadEventData = event.getData();
+ /**
+ * Loop over the uncompressed events in the
transaction payload event and add the table map
+ * event in the map of table events
+ **/
+ for (Event uncompressedEvent :
transactionPayloadEventData.getUncompressedEvents()) {
+ if (uncompressedEvent.getHeader().getEventType()
== EventType.TABLE_MAP && uncompressedEvent.getData() != null) {
+ TableMapEventData tableMapEvent =
uncompressedEvent.getData();
+
tableMapEventByTableId.put(tableMapEvent.getTableId(), tableMapEvent);
+ }
+ }
+ }
+ // Clean cache on rotate event to prevent it from growing
indefinitely.
+ if (event.getHeader().getEventType() == EventType.ROTATE) {
+ tableMapEventByTableId.clear();
+ }
+ return event;
+ } catch (EventDataDeserializationException ex) {
+ if (ex.getCause() instanceof IOException) {
+ throw ex;
+ }
+ EventHeaderV4 header = new EventHeaderV4();
+ header.setEventType(EventType.INCIDENT);
+ header.setTimestamp(ex.getEventHeader().getTimestamp());
+ header.setServerId(ex.getEventHeader().getServerId());
+ if (ex.getEventHeader() instanceof EventHeaderV4) {
+ header.setEventLength(((EventHeaderV4)
ex.getEventHeader()).getEventLength());
+ header.setNextPosition(((EventHeaderV4)
ex.getEventHeader()).getNextPosition());
+ header.setFlags(((EventHeaderV4)
ex.getEventHeader()).getFlags());
+ }
+ EventData data = new
EventDataDeserializationExceptionData(ex);
+ return new Event(header, data);
+ }
+ }
+ };
+
+ eventDeserializer.setEventDataDeserializer(EventType.WRITE_ROWS, new
WriteRowsEventMeshDeserializer(tableMapEventByTableId));
+ eventDeserializer.setEventDataDeserializer(EventType.UPDATE_ROWS, new
UpdateRowsEventMeshDeserializer(tableMapEventByTableId));
+ eventDeserializer.setEventDataDeserializer(EventType.DELETE_ROWS, new
DeleteRowsEventMeshDeserializer(tableMapEventByTableId));
+ eventDeserializer.setEventDataDeserializer(EventType.EXT_WRITE_ROWS,
+ new
WriteRowsEventMeshDeserializer(tableMapEventByTableId).setMayContainExtraInformation(true));
+ eventDeserializer.setEventDataDeserializer(EventType.EXT_UPDATE_ROWS,
+ new
UpdateRowsEventMeshDeserializer(tableMapEventByTableId).setMayContainExtraInformation(true));
+ eventDeserializer.setEventDataDeserializer(EventType.EXT_DELETE_ROWS,
+ new
DeleteRowsEventMeshDeserializer(tableMapEventByTableId).setMayContainExtraInformation(true));
+
+ // Set the event deserializer for the MySQL client
+ client.setEventDeserializer(eventDeserializer);
+ // Register an event listener for the MySQL client
+ client.registerEventListener((event) ->
eventMeshMysqlEventListener(event, context));
+ // Register a lifecycle listener for the MySQL client
+ client.registerLifecycleListener(new LifecycleListener() {
+ @Override
+ public void onConnect(BinaryLogClient client) {
+ log.info("Client connect MySQL Server success");
+ }
+
+ @Override
+ public void onCommunicationFailure(BinaryLogClient client,
Exception ex) {
+ log.error("Communicate with mysql error", ex);
+ }
+
+ @Override
+ public void onEventDeserializationFailure(BinaryLogClient client,
Exception ex) {
+ log.error("Event deserialization failure", ex);
+ }
+
+ @Override
+ public void onDisconnect(BinaryLogClient client) {
+ log.info("Disconnect Mysql");
+ }
+ });
+
+ //Register custom event handlers...
+ eventHandlers.put(EventType.STOP, event -> handleStopEvent(context,
event));
+ eventHandlers.put(EventType.HEARTBEAT, event ->
handleHeartbeatEvent(context, event));
+ eventHandlers.put(EventType.INCIDENT, event ->
handleServerIncident(context, event));
+ eventHandlers.put(EventType.ROTATE, event ->
handleRotateEvent(context, event));
+ eventHandlers.put(EventType.TABLE_MAP, event ->
handleTableMapEvent(context, event));
+ eventHandlers.put(EventType.QUERY, event -> handleQueryEvent(context,
event));
+ eventHandlers.put(EventType.TRANSACTION_PAYLOAD, event ->
handleTransactionPayload(context, event));
+
+ //Used to support 5.1.16 - mysql-trunk
+ eventHandlers.put(EventType.WRITE_ROWS, event ->
handleInsertEvent(context, event));
+ eventHandlers.put(EventType.UPDATE_ROWS, event ->
handleUpdateEvent(context, event));
+ eventHandlers.put(EventType.DELETE_ROWS, event ->
handleDeleteEvent(context, event));
+
+ // Used in case of RBR (5.1.18+).
+ eventHandlers.put(EventType.EXT_WRITE_ROWS, event ->
handleInsertEvent(context, event));
+ eventHandlers.put(EventType.EXT_UPDATE_ROWS, event ->
handleUpdateEvent(context, event));
+ eventHandlers.put(EventType.EXT_DELETE_ROWS, event ->
handleDeleteEvent(context, event));
+
+ eventHandlers.put(EventType.VIEW_CHANGE, (event) ->
handleViewChangeEvent(context, event));
+ eventHandlers.put(EventType.XA_PREPARE, (event) ->
handleXAPrepareTransactionEvent(context, event));
+ eventHandlers.put(EventType.XID, (event) ->
handleTransactionCompletionEvent(context, event));
+ }
+
+ public EventMeshGtidSet filterGtidSet(MysqlJdbcContext offsetContext,
EventMeshGtidSet availableServerEventMeshGtidSet,
+ EventMeshGtidSet purgedServerGtid) {
+ String gtidStr = offsetContext.getGtidSet();
+ if (gtidStr == null) {
+ return null;
+ }
+
+ EventMeshGtidSet filteredEventMeshGtidSet = new
EventMeshGtidSet(gtidStr);
+
+ final EventMeshGtidSet knownEventMeshGtidSet =
filteredEventMeshGtidSet;
+ final EventMeshGtidSet relevantAvailableServerEventMeshGtidSet =
availableServerEventMeshGtidSet;
+
+ EventMeshGtidSet mergedEventMeshGtidSet =
relevantAvailableServerEventMeshGtidSet
+ .retainAll(uuid -> knownEventMeshGtidSet.forServerWithId(uuid) !=
null)
+ .with(purgedServerGtid)
+ .with(filteredEventMeshGtidSet);
+
+ return mergedEventMeshGtidSet;
+ }
+
+
+ /**
+ * handles events from the MySQL
+ *
+ * @param event mysql binlog event
+ * @param context mysql context
+ */
+ private void eventMeshMysqlEventListener(Event event, MysqlJdbcContext
context) {
+
+ if (null == event) {
+ return;
+ }
+ final EventHeader eventHeader = event.getHeader();
+ final EventType eventType = eventHeader.getEventType();
+ /**
+ * @see <a
href="https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_replication_binlog_event.html#sect_protocol_replication_event_rotate";>ROTATE_EVENT</a>
+ * +----------------------------------------------------+
+ * | ROTATE_EVENT |
+ * +----------------------------------------------------+
+ * | Event Header | Position Info | Filename|
+ * +----------------------------------------------------+
+ * The rotate event is added to the binlog as last event to tell the
reader what binlog to request next.
+ */
+ if (eventType == EventType.ROTATE) {
+ RotateEventData rotateEventData = unwrapData(event);
+ context.setBinlogStartPoint(rotateEventData.getBinlogFilename(),
rotateEventData.getBinlogPosition());
+ } else if (eventHeader instanceof EventHeaderV4) {
+ EventHeaderV4 eventHeaderV4 = (EventHeaderV4) eventHeader;
+ context.setEventPosition(eventHeaderV4.getPosition(),
eventHeaderV4.getEventLength());
+ }
+ if (eventType == EventType.HEARTBEAT) {
+ return;
+ }
+ try {
+ eventQueue.put(event);
+ } catch (InterruptedException e) {
+ log.warn("Put event to queue error", e);
+ }
+ context.complete();
+ }
+
+
+ @Override
+ public String getThreadName() {
+ return "MySQL-CdcEngine";
+ }
+
+ /**
+ * Returns the name of the CDC Engine.
+ *
+ * @return String representing the name of the CDC Engine.
+ */
+ @Override
+ public String getCdcEngineName() {
+ return "MySQL CDC Engine";
+ }
+
+ @Override
+ public void close() throws Exception {
+ client.disconnect();
+ }
+
+ /**
+ * When an object implementing interface <code>Runnable</code> is used to
create a thread, starting the thread causes the object's
+ * <code>run</code> method to be called in that separately executing
+ * thread.
+ * <p>
+ * The general contract of the method <code>run</code> is that it may take
any action whatsoever.
+ *
+ * @see Thread#run()
+ */
+ @Override
+ public void run() {
+ enableGtidHandle();
+ do {
+ try {
+ client.connect(TimeUnit.SECONDS.toMillis(5));
+ } catch (IOException | TimeoutException e) {
+ log.error("Binary log client connect to mysql server error,
The connection will be retried in three seconds", e);
+ await(3, TimeUnit.SECONDS);
+ }
+ } while (!client.isConnected());
+
+ while (isRunning) {
+ Event event = null;
+ try {
+ event = eventQueue.poll(5, TimeUnit.SECONDS);
+ if (null == event) {
+ continue;
+ }
+ eventHandlers.getOrDefault(event.getHeader().getEventType(),
ignore -> ignoreEvent(context, ignore)).accept(event);
+ } catch (Exception e) {
+ if (event != null) {
+ log.warn("Handle EventType={} error",
event.getHeader().getEventType(), e);
+ }
+ }
+ }
+ }
+
+ private void enableGtidHandle() {
+ //Query whether mysql supports GTID
+ if (this.connection.enableGTID()) {
+ eventHandlers.put(EventType.GTID, event ->
handleGtidEvent(context, event));
+
+ //Query GtidSet from the MySQL Server
+ String availableServerGtid = this.connection.executedGTID();
+ EventMeshGtidSet executedEventMeshGtidSet = new
EventMeshGtidSet(availableServerGtid);
+
+ //Get purged GTID
+ String purgedServerGtid = this.connection.purgedGTID();
+ EventMeshGtidSet purgedServerEventMeshGtidSet = new
EventMeshGtidSet(purgedServerGtid);
+
+ EventMeshGtidSet filteredEventMeshGtidSet = filterGtidSet(context,
executedEventMeshGtidSet, purgedServerEventMeshGtidSet);
+ if (null != filteredEventMeshGtidSet) {
+ client.setGtidSet(filteredEventMeshGtidSet.toString());
+
this.context.completedGtidSet(filteredEventMeshGtidSet.toString());
+ localGtidSet = new
com.github.shyiko.mysql.binlog.GtidSet(filteredEventMeshGtidSet.toString());
+ } else {
+
client.setBinlogFilename(this.context.getSourceInfo().getCurrentBinlogFileName());
+
client.setBinlogPosition(this.context.getSourceInfo().getCurrentBinlogPosition());
+ localGtidSet = new com.github.shyiko.mysql.binlog.GtidSet("");
+ }
+ } else {
+
client.setBinlogFilename(this.context.getSourceInfo().getCurrentBinlogFileName());
+
client.setBinlogPosition(this.context.getSourceInfo().getCurrentBinlogPosition());
+ }
+ }
+
+ /**
+ * Handles the STOP_EVENT
+ *
+ * @param context the MySQL context
+ * @param event the event to be handled
+ */
+ protected void handleStopEvent(MysqlJdbcContext context, Event event) {
+ //The purpose of STOP_EVENT is to inform MySQL that the slave or
replication client has reached the end of the binary log
+ // and no new events will be generated. When the slave receives a
STOP_EVENT, it can take appropriate actions based on its needs,
+ // such as closing the connection to the master server or reconnecting
to obtain a new binary log file.
+ log.debug("Replication client has reached the end of the binary log:
{}", event);
+ }
+
+ /**
+ * Handles the HEARTBEAT_EVENT
+ *
+ * @param context the MySQL context
+ * @param event the event to be handled
+ */
+ protected void handleHeartbeatEvent(MysqlJdbcContext context, Event event)
{
+ if (log.isDebugEnabled()) {
+ log.debug("Replication client handle {}",
event.getHeader().getEventType());
+ }
+ }
+
+ /**
+ * Handles the <a
href="https://dev.mysql.com/doc/dev/mysql-server/latest/classbinary__log_1_1Incident__event.html";>INCIDENT_EVENT</a>
+ *
+ * @param context the MySQL context
+ * @param event the event to be handled
+ */
+ protected void handleServerIncident(MysqlJdbcContext context, Event event)
{
+
+ final EventData eventData = event.getData();
+ if (eventData instanceof EventDataDeserializationExceptionData) {
+ log.error("Server incident: {}", event);
+ }
+
+ }
+
+ /**
+ * Handles the ROTATE_EVENT
+ *
+ * @param context the MySQL context
+ * @param event the event to be handled
+ */
+ protected void handleRotateEvent(MysqlJdbcContext context, Event event) {
+ RotateEventData eventData = unwrapData(event);
+ assert eventData != null;
+ tableIdMap.clear();
+ }
+
+ @SuppressWarnings("unchecked")
+ protected <T extends EventData> T unwrapData(Event event) {
+ EventData eventData = event.getData();
+ if (eventData instanceof EventDeserializer.EventDataWrapper) {
+ eventData = ((EventDeserializer.EventDataWrapper)
eventData).getInternal();
+ }
+ return (T) eventData;
+ }
+
+ /**
+ * Handles the TABLE_MAP_EVENT Format @see <a
+ *
href="https://dev.mysql.com/doc/dev/mysql-server/latest/classbinary__log_1_1Table__map__event.html";>TABLE_MAP_EVENT</a>
+ *
+ * @param context the MySQL context
+ * @param event the event to be handled
+ */
+ protected void handleTableMapEvent(MysqlJdbcContext context, Event event) {
+ TableMapEventData tableMapEventData = event.getData();
+ final long tableId = tableMapEventData.getTableId();
+ final String tableName = tableMapEventData.getTable();
+ final String database = tableMapEventData.getDatabase();
+ tableIdMap.put(tableId, new TableId(database, null, tableName));
+ }
+
+ /**
+ * Handles the QUERY_EVENT
+ *
+ * @param context mysql context
+ * @param event query event
+ */
+ protected void handleQueryEvent(MysqlJdbcContext context, Event event) {
+ QueryEventData queryEventData = unwrapData(event);
+ final String sql = queryEventData.getSql().trim();
+ log.debug("Received query event SQL:{}", sql);
+ if (StringUtils.equalsIgnoreCase("BEGIN", sql)) {
+ //start transaction
+ context.startTransaction();
+ return;
+ }
+
+ if (StringUtils.equalsIgnoreCase("COMMIT", sql)) {
+ context.commitTransaction();
+ return;
+ }
+
+ if (StringUtils.startsWithIgnoreCase("XA", sql)) {
+ //TODO: next version support
+
+ return;
+ }
+ String sqlBegin = sql.substring(0, 6).toUpperCase();
+ if (StringUtils.startsWithAny(sqlBegin, "INSERT", "UPDATE", "DELETE"))
{
+ LogUtils.warn(log, "Received DML '[SQL={}]' for processing, binlog
probably contains events generated with statement", sql);
+ return;
+ }
+
+ //set current parse database to Ddl parser
+ ddlParser.setCurrentDatabase(queryEventData.getDatabase());
+ ddlParser.parse(sql, this::handleDdlEvent);
+ }
+
+ private void
handleDdlEvent(org.apache.eventmesh.connector.jdbc.event.Event event) {
+ if (event == null) {
+ return;
+ }
+
event.getJdbcConnectData().getPayload().ofSourceMateData().setSnapshot(false);
+ consumers.stream().forEach(consumer -> consumer.accept(event));
+ }
+
+ /**
+ * Handles the TRANSACTION_PAYLOAD_EVENT
+ * <p>
+ * "binlog_transaction_compression" is a new feature introduced in MySQL
8.0.23, used for compressing transactions in the binary log (binlog).
+ * This event is a wrapper event and encloses many other events.It is
mostly used for carrying compressed payloads as its content can be
+ * compressed, in which case, its metadata shall contain information about
the compression metadata as well.
+ *
+ * </p>
+ *
+ * @param context the MySQL context
+ * @param event the event to be handled
+ */
+ protected void handleTransactionPayload(MysqlJdbcContext context, Event
event) {
+
+ TransactionPayloadEventData transactionPayloadEventData =
event.getData();
+ //unpack Event and handle
+ ArrayList<Event> uncompressedEvents =
transactionPayloadEventData.getUncompressedEvents();
+ for (Event uncompressedEvent : uncompressedEvents) {
+ final EventType eventType =
uncompressedEvent.getHeader().getEventType();
+ eventHandlers.getOrDefault(eventType, et -> ignoreEvent(context,
et)).accept(uncompressedEvent);
+ }
+ }
+
+ /**
+ * Handles the insert event.
+ *
+ * @param context The MySQL context.
+ * @param event The insert event.
+ */
+ protected void handleInsertEvent(MysqlJdbcContext context, Event event) {
+ WriteRowsEventData writeRowsEventData = unwrapData(event);
+ log.debug("Received Write rows event, TableId={}",
writeRowsEventData.getTableId());
+ long tableNumber = writeRowsEventData.getTableId();
+ TableId tableId = tableIdMap.get(tableNumber);
+
+ if (!getHandledTables().contains(tableId)) {
+ log.warn("Write rows-Table {} is excluded", tableId);
+ return;
+ }
+ MysqlSourceMateData sourceMateData = buildMysqlSourceMateData(context,
event, tableId);
+ List<Serializable[]> insertRows = writeRowsEventData.getRows();
+ if (CollectionUtils.isEmpty(insertRows)) {
+ return;
+ }
+ List<Pair<Pair<Serializable[], BitSet>, Pair<Serializable[], BitSet>>>
rows = new ArrayList<>();
+ for (Serializable[] row : insertRows) {
+ Pair<Serializable[], BitSet> item = new Pair<>(row,
writeRowsEventData.getIncludedColumns());
+ rows.add(new Pair<>(null, item));
+ }
+ handleCdcDmlData(context, sourceMateData, tableId, rows,
CdcDmlType.INSERT);
+ }
+
+ private MysqlSourceMateData buildMysqlSourceMateData(MysqlJdbcContext
context, Event event, TableId tableId) {
+ MysqlSourceMateData sourceMateData = MysqlSourceMateData.newBuilder()
+ .name(sourceConnectorConfig.getName())
+ .withTableId(tableId)
+ .serverId(sourceConnectorConfig.getMysqlConfig().getServerId())
+ .binlogFile(context.getSourceInfo().getCurrentBinlogFileName())
+ .position(((EventHeaderV4) event.getHeader()).getPosition())
+ .build();
+ return sourceMateData;
+ }
+
+ private enum CdcDmlType {
+ INSERT,
+ UPDATE,
+ DELETE
+ }
+
+ private GeneralDataChangeEvent buildEvent(CdcDmlType type, TableId
tableId) {
+ switch (type) {
+ case UPDATE:
+ return new UpdateDataEvent(tableId);
+ case INSERT:
+ return new InsertDataEvent(tableId);
+ case DELETE:
+ return new DeleteDataEvent(tableId);
+ default:
+ return null;
+ }
+ }
+
+ private void handleCdcDmlData(MysqlJdbcContext context,
MysqlSourceMateData sourceMateData, TableId tableId,
+ List<Pair<Pair<Serializable[], BitSet>, Pair<Serializable[], BitSet>>>
rows, CdcDmlType type) {
+
+ TableSchema tableSchema =
context.getCatalogTableSet().getTableSchema(tableId);
+ Map<Integer, ? extends Column> orderColumnMap =
tableSchema.getOrderColumnMap();
+ List<? extends Column> columns = tableSchema.getColumns();
+ List<Field> fields = null;
+ Builder builder = DataChanges.newBuilder();
+ if (CollectionUtils.isNotEmpty(columns)) {
+ fields = columns.stream()
+ .map(col -> new Field(col.getDataType().getName(),
col.isNotNull(), col.getName(), tableId.toString()))
+ .collect(Collectors.toList());
+ }
+ int columnsSize = orderColumnMap.size();
+ for (Pair<Pair<Serializable[], BitSet>, Pair<Serializable[], BitSet>>
pair : rows) {
+ GeneralDataChangeEvent dataEvent = buildEvent(type, tableId);
+ Payload payload = dataEvent.getJdbcConnectData().getPayload();
+ Schema schema = new Schema();
+ Pair<Serializable[], BitSet> beforePair =
Optional.ofNullable(pair.getLeft()).orElse(new Pair<>());
+ Serializable[] beforeRows = beforePair.getLeft();
+ if (null != beforeRows && beforeRows.length != 0) {
+ BitSet includedColumns = beforePair.getRight();
+ Map<String, Object> beforeValues = new
HashMap<>(beforeRows.length);
+ for (int index = 0; index < columnsSize; ++index) {
+ //Filter out empty fields
+ if (!includedColumns.get(index)) {
+ continue;
+ }
+ beforeValues.put(orderColumnMap.get(index + 1).getName(),
beforeRows[index]);
+ }
+ builder.withBefore(beforeValues);
+ Field beforeField = new
Field().withField(Payload.BEFORE_FIELD).withType("field").withName("payload.before").withRequired(false);
+ beforeField.withRequired(true).withFields(fields);
+ schema.add(beforeField);
+ }
+
+ Pair<Serializable[], BitSet> afterPair =
Optional.ofNullable(pair.getRight()).orElse(new Pair<>());
+ Serializable[] afterRows = afterPair.getLeft();
+ if (null != afterRows && afterRows.length != 0) {
+ BitSet includedColumns = afterPair.getRight();
+ Map<String, Object> afterValues = new
HashMap<>(afterRows.length);
+ for (int index = 0; index < columnsSize; ++index) {
+ //Filter out empty fields
+ if (!includedColumns.get(index)) {
+ continue;
+ }
+ afterValues.put(orderColumnMap.get(index + 1).getName(),
afterRows[index]);
+ }
+ builder.withBefore(afterValues);
+ Field afterField = new
Field().withField(Payload.AFTER_FIELD).withType("field").withName("payload.after").withRequired(false);
+ afterField.withRequired(true).withFields(fields);
+ schema.add(afterField);
+ }
+
payload.withSource(sourceMateData).withDataChanges(builder.build());
+ dataEvent.getJdbcConnectData().setSchema(schema);
+ consumers.stream().forEach(consumer -> consumer.accept(dataEvent));
+
+ }
+ }
+
+ /**
+ * Handles the update event.
+ *
+ * @param context The MySQL context.
+ * @param event The update event.
+ */
+ protected void handleUpdateEvent(MysqlJdbcContext context, Event event) {
+ UpdateRowsEventData updateRowsEventData = unwrapData(event);
+ log.debug("Received Update rows event, Update table is {}",
tableIdMap.get(updateRowsEventData.getTableId()));
+ long id = updateRowsEventData.getTableId();
+ TableId tableId = tableIdMap.get(id);
+ if (!getHandledTables().contains(tableId)) {
+ log.debug("Update rows-Table {} is excluded", tableId);
+ return;
+ }
+ MysqlSourceMateData sourceMateData = buildMysqlSourceMateData(context,
event, tableId);
+ List<Entry<Serializable[], Serializable[]>> updateRows =
updateRowsEventData.getRows();
+ if (CollectionUtils.isEmpty(updateRows)) {
+ return;
+ }
+ List<Pair<Pair<Serializable[], BitSet>, Pair<Serializable[], BitSet>>>
rows = new ArrayList<>();
+ for (Entry<Serializable[], Serializable[]> row : updateRows) {
+ Pair<Serializable[], BitSet> before = new Pair<>(row.getKey(),
updateRowsEventData.getIncludedColumnsBeforeUpdate());
+ Pair<Serializable[], BitSet> after = new Pair<>(row.getValue(),
updateRowsEventData.getIncludedColumns());
+ rows.add(new Pair<>(before, after));
+ }
+ handleCdcDmlData(context, sourceMateData, tableId, rows,
CdcDmlType.UPDATE);
+ }
+
+ /**
+ * Handles the delete event.
+ *
+ * @param context The MySQL context.
+ * @param event The delete event.
+ */
+ protected void handleDeleteEvent(MysqlJdbcContext context, Event event) {
+ DeleteRowsEventData deleteRowsEventData = unwrapData(event);
+ log.debug("Received Delete rows event, Delete table is {}",
tableIdMap.get(deleteRowsEventData.getTableId()));
+ long id = deleteRowsEventData.getTableId();
+ TableId tableId = tableIdMap.get(id);
+
+ if (!getHandledTables().contains(tableId)) {
+ log.debug("Update rows-Table {} is excluded", tableId);
+ return;
+ }
+ MysqlSourceMateData sourceMateData = buildMysqlSourceMateData(context,
event, tableId);
+ List<Serializable[]> deleteRows = deleteRowsEventData.getRows();
+ if (CollectionUtils.isEmpty(deleteRows)) {
+ return;
+ }
+ List<Pair<Pair<Serializable[], BitSet>, Pair<Serializable[], BitSet>>>
rows = new ArrayList<>();
+ for (Serializable[] row : deleteRows) {
+ Pair<Serializable[], BitSet> item = new Pair<>(row,
deleteRowsEventData.getIncludedColumns());
+ rows.add(new Pair<>(item, null));
+ }
+ handleCdcDmlData(context, sourceMateData, tableId, rows,
CdcDmlType.DELETE);
+ }
+
+ /**
+ * Handles the GTID event.
+ *
+ * @param context The MySQL context.
+ * @param event The GTID event.
+ */
+ protected void handleGtidEvent(MysqlJdbcContext context, Event event) {
+ GtidEventData gtidEvent = unwrapData(event);
+ String gtid = gtidEvent.getMySqlGtid().toString();
+ LogUtils.debug(log, "Received GTID event: {}", gtid);
+ localGtidSet.add(gtid);
+ context.beginGtid(gtid);
+ }
+
+ /**
+ * Handles the view change event.
+ *
+ * @param context The MySQL context.
+ * @param event The view change event.
+ */
+ protected void handleViewChangeEvent(MysqlJdbcContext context, Event
event) {
+ // TODO: Add support for handling view change event
+ }
+
+ /**
+ * Handles the XA prepare transaction event.
+ *
+ * @param context The MySQL context.
+ * @param event The XA prepare transaction event.
+ */
+ protected void handleXAPrepareTransactionEvent(MysqlJdbcContext context,
Event event) {
+ // TODO: Add support for handling XA prepare transaction event
+ }
+
+ /**
+ * Handles the transaction completion event.
+ *
+ * @param context The MySQL context.
+ * @param event The transaction completion event.
+ */
+ protected void handleTransactionCompletionEvent(MysqlJdbcContext context,
Event event) {
+ XidEventData xidEventData = unwrapData(event);
+ log.debug("Received XID event, Xid={}", xidEventData.getXid());
+ context.commitTransaction();
+ }
+
+ /**
+ * Default handler that ignores events.
+ *
+ * @param context The MySQL context.
+ * @param event The event to be ignored.
+ */
+ protected void ignoreEvent(MysqlJdbcContext context, Event event) {
+ log.debug("Ignoring event due to missing handler: {}", event);
+ }
+
+ @Override
+ public void registerCdcEventConsumer(EventConsumer consumer) {
+ if (consumer == null) {
+ return;
+ }
+ consumers.add(consumer);
+ }
+
+ @Override
+ protected Set<String> defaultExcludeDatabase() {
+ return MysqlConstants.DEFAULT_EXCLUDE_DATABASE;
+ }
+
+ @Override
+ protected MysqlAntlr4DdlParser getDdlParser() {
+ return ddlParser;
+ }
+
+ @Override
+ public void setContext(MysqlJdbcContext context) {
+ if (context == null) {
+ context = MysqlJdbcContext.initialize(this.jdbcSourceConfig);
+ }
+ this.context = context;
+ }
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngineFactory.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngineFactory.java
new file mode 100644
index 000000000..60c3c7b7c
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/MysqlCdcEngineFactory.java
@@ -0,0 +1,54 @@
+/*
+ * 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.eventmesh.connector.jdbc.source.dialect.cdc.mysql;
+
+import org.apache.eventmesh.connector.jdbc.DatabaseDialect;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.cdc.AbstractCdcEngineFactory;
+import org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngine;
+import org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngineFactory;
+import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlDatabaseDialect;
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
+
+import org.apache.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class MysqlCdcEngineFactory extends AbstractCdcEngineFactory implements
CdcEngineFactory {
+
+ @Override
+ public boolean acceptJdbcProtocol(String url) {
+ if (StringUtils.isBlank(url)) {
+ return true;
+ }
+ return url.toLowerCase().startsWith("jdbc:mysql");
+ }
+
+ /**
+ * Creates a CDC engine based on the provided source configuration and
database dialect.
+ *
+ * @param config the source configuration for the CDC engine
+ * @param databaseDialect the database dialect for the CDC engine
+ * @return the created CDC engine
+ */
+ @Override
+ public CdcEngine createCdcEngine(SourceConfig config, DatabaseDialect
databaseDialect) {
+ return new MysqlCdcEngine(config, (MysqlDatabaseDialect)
databaseDialect);
+ }
+
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/RowDeserializers.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/RowDeserializers.java
new file mode 100644
index 000000000..e4b1fde87
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/mysql/RowDeserializers.java
@@ -0,0 +1,452 @@
+/*
+ * 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.eventmesh.connector.jdbc.source.dialect.cdc.mysql;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.Year;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.util.Map;
+
+import com.github.shyiko.mysql.binlog.event.TableMapEventData;
+import
com.github.shyiko.mysql.binlog.event.deserialization.DeleteRowsEventDataDeserializer;
+import
com.github.shyiko.mysql.binlog.event.deserialization.UpdateRowsEventDataDeserializer;
+import
com.github.shyiko.mysql.binlog.event.deserialization.WriteRowsEventDataDeserializer;
+import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Custom deserializers for the MySQL Binlog Client.MySQL Binlog Client row
deserializers convert MySQL raw row data into {@link java.sql.Date},
+ * {@link java.sql.Time}, and {@link java.sql.Timestamp} values using {@link
java.util.Calendar} instances. EventMesh convert the raw MySQL row values
+ * directly into {@link LocalDate}, {@link LocalTime}, {@link LocalDateTime},
and {@link java.time.OffsetDateTime}.
+ */
+@Slf4j
+public class RowDeserializers {
+
+ public static class WriteRowsEventMeshDeserializer extends
WriteRowsEventDataDeserializer {
+
+ public WriteRowsEventMeshDeserializer(Map<Long, TableMapEventData>
tableMapEventByTableId) {
+ super(tableMapEventByTableId);
+ }
+
+ @Override
+ protected Serializable deserializeString(int length,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeString(length, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeVarString(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeVarString(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDate(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeDate(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDatetime(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeDatetime(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDatetimeV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeDatetimeV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimeV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeTimeV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTime(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeTime(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimestamp(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeTimestamp(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimestampV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeTimestampV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeYear(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeYear(inputStream);
+ }
+ }
+
+ public static class UpdateRowsEventMeshDeserializer extends
UpdateRowsEventDataDeserializer {
+
+ public UpdateRowsEventMeshDeserializer(Map<Long, TableMapEventData>
tableMapEventByTableId) {
+ super(tableMapEventByTableId);
+ }
+
+ @Override
+ protected Serializable deserializeString(int length,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeString(length, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeVarString(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeVarString(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDate(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeDate(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDatetime(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeDatetime(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDatetimeV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeDatetimeV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimeV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeTimeV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTime(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeTime(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimestamp(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeTimestamp(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimestampV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeTimestampV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeYear(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeYear(inputStream);
+ }
+ }
+
+ public static class DeleteRowsEventMeshDeserializer extends
DeleteRowsEventDataDeserializer {
+
+ public DeleteRowsEventMeshDeserializer(Map<Long, TableMapEventData>
tableMapEventByTableId) {
+ super(tableMapEventByTableId);
+ }
+
+ @Override
+ protected Serializable deserializeString(int length,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeString(length, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeVarString(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeVarString(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDate(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeDate(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDatetime(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeDatetime(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeDatetimeV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeDatetimeV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimeV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeTimeV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTime(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeTime(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimestamp(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeTimestamp(inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeTimestampV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ return RowDeserializers.deserializeTimestampV2(meta, inputStream);
+ }
+
+ @Override
+ protected Serializable deserializeYear(ByteArrayInputStream
inputStream) throws IOException {
+ return RowDeserializers.deserializeYear(inputStream);
+ }
+ }
+
+ protected static Serializable deserializeTimestamp(ByteArrayInputStream
inputStream) throws IOException {
+ long epochSecond = inputStream.readLong(4);
+ int nanoSeconds = 0; // no fractional seconds
+ return ZonedDateTime.ofInstant(Instant.ofEpochSecond(epochSecond,
nanoSeconds), ZoneOffset.UTC);
+ }
+
+ protected static Serializable deserializeTimestampV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ long epochSecond = bigEndianLong(inputStream.read(4), 0, 4);
+ int nanoSeconds = deserializeFractionalSecondsInNanos(meta,
inputStream);
+ return ZonedDateTime.ofInstant(Instant.ofEpochSecond(epochSecond,
nanoSeconds), ZoneOffset.UTC);
+ }
+
+ protected static Serializable deserializeYear(ByteArrayInputStream
inputStream) throws IOException {
+ return Year.of(1900 + inputStream.readInteger(1));
+ }
+
+ /**
+ * Deserializes a string from the given input stream. Since the charset is
not present in the binary log, it is impossible to distinguish between
+ * CHAR and BINARY types. Therefore, the method returns a byte array
instead of a String.
+ *
+ * @param length The length of the string.
+ * @param inputStream The input stream from which to read the string.
+ * @return A byte array representing the deserialized string.
+ * @throws IOException If an I/O error occurs while reading the input
stream.
+ */
+ private static Serializable deserializeString(int length,
ByteArrayInputStream inputStream) throws IOException {
+ // charset is not present in the binary log (meaning there is no way
to distinguish between CHAR / BINARY)
+ // as a result - return byte[] instead of an actual String
+ int stringLength = length < 256 ? inputStream.readInteger(1) :
inputStream.readInteger(2);
+ return inputStream.read(stringLength);
+ }
+
+
+ private static Serializable deserializeVarString(int meta,
ByteArrayInputStream inputStream) throws IOException {
+ int varcharLength = meta < 256 ? inputStream.readInteger(1) :
inputStream.readInteger(2);
+ return inputStream.read(varcharLength);
+ }
+
+ private static Serializable deserializeDate(ByteArrayInputStream
inputStream) throws IOException {
+ int value = inputStream.readInteger(3);
+ int day = value % 32;
+ value >>>= 5;
+ int month = value % 16;
+ int year = value >> 4;
+ // https://dev.mysql.com/doc/refman/8.0/en/datetime.html
+ if (year == 0 || month == 0 || day == 0) {
+ return null;
+ }
+ return LocalDate.of(year, month, day);
+ }
+
+ protected static Serializable deserializeDatetime(ByteArrayInputStream
inputStream) throws IOException {
+ int[] split = split(inputStream.readLong(8), 100, 6);
+ int year = split[5];
+ int month = split[4]; // 1-based month number
+ int day = split[3]; // 1-based day of the month
+ int hours = split[2];
+ int minutes = split[1];
+ int seconds = split[0];
+ int nanoOfSecond = 0; // This version does not support fractional
seconds
+ if (year == 0 || month == 0 || day == 0) {
+ return null;
+ }
+ return LocalDateTime.of(year, month, day, hours, minutes, seconds,
nanoOfSecond);
+ }
+
+ private static final int MASK_10_BITS = (1 << 10) - 1;
+ private static final int MASK_6_BITS = (1 << 6) - 1;
+
+ protected static Serializable deserializeTimeV2(int meta,
ByteArrayInputStream inputStream) throws IOException {
+
+ /**
+ * Binary Format for TIME in MySQL binlog:
+ * <a
href="https://dev.mysql.com/doc/dev/mysql-server/latest/group__MY__TIME.html#time_low_level_rep";>TIME</a>
+ *
+------+---------------+----------------------------------------------------+
+ * | Bits | Field | Value Range
|
+ *
+------+---------------+----------------------------------------------------+
+ * | 1 | sign | (Used for sign, when on disk)
|
+ * | 1 | unused | (Reserved for wider hour range, e.g. for
intervals)|
+ * | 10 | hour | (0-838)
|
+ * | 6 | minute | (0-59)
|
+ * | 6 | second | (0-59)
|
+ * | 24 | microseconds | (0-999999)
|
+ *
+------+---------------+----------------------------------------------------+
+ *
+ * + fractional-seconds storage (size depends on meta)
+ */
+ long time = bigEndianLong(inputStream.read(3), 0, 3);
+ boolean isNegative = bitSlice(time, 0, 1, 24) == 0;
+ int hours = bitSlice(time, 2, 10, 24);
+ int minutes = bitSlice(time, 12, 6, 24);
+ int seconds = bitSlice(time, 18, 6, 24);
+ int nanoSeconds;
+ if (isNegative) { // mysql binary arithmetic for negative encoded
values
+ hours = ~hours & MASK_10_BITS;
+ hours = hours & ~(1 << 10); // unset sign bit
+ minutes = ~minutes & MASK_6_BITS;
+ minutes = minutes & ~(1 << 6); // unset sign bit
+ seconds = ~seconds & MASK_6_BITS;
+ seconds = seconds & ~(1 << 6); // unset sign bit
+ nanoSeconds = deserializeFractionalSecondsInNanosNegative(meta,
inputStream);
+ if (nanoSeconds == 0 && seconds < 59) { // weird java Duration
behavior
+ ++seconds;
+ }
+ hours = -hours;
+ minutes = -minutes;
+ seconds = -seconds;
+ nanoSeconds = -nanoSeconds;
+ } else {
+ nanoSeconds = deserializeFractionalSecondsInNanos(meta,
inputStream);
+ }
+
+ return LocalTime.of(hours, minutes, seconds, nanoSeconds);
+ }
+
+ protected static Serializable deserializeTime(ByteArrayInputStream
inputStream) throws IOException {
+ // Times are stored as an integer as `HHMMSS`, so we need to split out
the digits ...
+ int value = inputStream.readInteger(3);
+ int[] split = split(value, 100, 3);
+ int hours = split[2];
+ int minutes = split[1];
+ int seconds = split[0];
+ return LocalTime.of(hours, minutes, seconds);
+ }
+
+ protected static int deserializeFractionalSecondsInNanosNegative(int fsp,
ByteArrayInputStream inputStream) throws IOException {
+ // Calculate the number of bytes to read, which is
+ // '1' when fsp=(1,2)
+ // '2' when fsp=(3,4) and
+ // '3' when fsp=(5,6)
+ int length = (fsp + 1) / 2;
+ if (length > 0) {
+ long fraction = bigEndianLong(inputStream.read(length), 0, length);
+ int maskBits = 0;
+ switch (length) { // mask bits according to field precision
+ case 1:
+ maskBits = 8;
+ break;
+ case 2:
+ maskBits = 15;
+ break;
+ case 3:
+ maskBits = 20;
+ break;
+ default:
+ break;
+ }
+ fraction = ~fraction & ((1 << maskBits) - 1);
+ fraction = (fraction & ~(1 << maskBits)) + 1; // unset sign bit
+ // Convert the fractional value (which has extra trailing digit
for fsp=1,3, and 5) to nanoseconds ...
+ return (int) (fraction / (0.0000001 * Math.pow(100, length - 1)));
+ }
+ return 0;
+ }
+
+ private static int bigEndianInteger(byte[] bytes, int offset, int length) {
+ int result = 0;
+ for (int i = offset; i < (offset + length); i++) {
+ byte b = bytes[i];
+ result = (result << 8) | (b >= 0 ? (int) b : (b + 256));
+ }
+ return result;
+ }
+
+ protected static Serializable deserializeDatetimeV2(int meta,
ByteArrayInputStream inputStream)
+ throws IOException {
+
+ /**
+ * <a href =
"https://dev.mysql.com/doc/dev/mysql-server/latest/group__MY__TIME.html";>DATETIME</a>
+ * 1 sign (used when on disk)
+ * 17 year*13+month (year 0-9999, month 0-12)
+ * 5 day (0-31)
+ * 5 hour (0-23)
+ * 6 minute (0-59)
+ * 6 second (0-59)
+ * 24 microseconds (0-999999)
+ *
+ * (5 bytes in total)
+ *
+ * + fractional-seconds storage (size depends on meta)
+ */
+ long datetime = bigEndianLong(inputStream.read(5), 0, 5);
+ int yearMonth = bitSlice(datetime, 1, 17, 40);
+ int year = yearMonth / 13;
+ int month = yearMonth % 13; // 1-based month number
+ int day = bitSlice(datetime, 18, 5, 40); // 1-based day of the month
+ int hours = bitSlice(datetime, 23, 5, 40);
+ int minutes = bitSlice(datetime, 28, 6, 40);
+ int seconds = bitSlice(datetime, 34, 6, 40);
+ int nanoOfSecond = deserializeFractionalSecondsInNanos(meta,
inputStream);
+ if (year == 0 || month == 0 || day == 0) {
+ return null;
+ }
+ return LocalDateTime.of(year, month, day, hours, minutes, seconds,
nanoOfSecond);
+ }
+
+ private static int[] split(long value, int divider, int length) {
+ int[] result = new int[length];
+ for (int i = 0; i < length - 1; i++) {
+ result[i] = (int) (value % divider);
+ value /= divider;
+ }
+ result[length - 1] = (int) value;
+ return result;
+ }
+
+ protected static long bigEndianLong(byte[] bytes, int offset, int length) {
+ long result = 0;
+ for (int i = offset; i < (offset + length); i++) {
+ byte b = bytes[i];
+ result = (result << 8) | (b >= 0 ? (int) b : (b + 256));
+ }
+ return result;
+ }
+
+ protected static int bitSlice(long value, int bitOffset, int numberOfBits,
int payloadSize) {
+ long result = value >> payloadSize - (bitOffset + numberOfBits);
+ return (int) (result & ((1 << numberOfBits) - 1));
+ }
+
+ protected static int deserializeFractionalSecondsInNanos(int fsp,
ByteArrayInputStream inputStream) throws IOException {
+ // Calculate the number of bytes to read, which is
+ // '1' when fsp=(1,2) -- 7
+ // '2' when fsp=(3,4) and -- 12
+ // '3' when fsp=(5,6) -- 21
+ int length = (fsp + 1) / 2;
+ if (length > 0) {
+ long fraction = bigEndianLong(inputStream.read(length), 0, length);
+ // Convert the fractional value (which has extra trailing digit
for fsp=1,3, and 5) to nanoseconds ...
+ return (int) (fraction / (0.0000001 * Math.pow(100, length - 1)));
+ }
+ return 0;
+ }
+
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/GtidSet.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/EventMeshGtidSet.java
similarity index 93%
rename from
eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/GtidSet.java
rename to
eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/EventMeshGtidSet.java
index a1cb8738e..481aedb3d 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/GtidSet.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/EventMeshGtidSet.java
@@ -32,19 +32,19 @@ import java.util.function.Predicate;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
-public final class GtidSet {
+public final class EventMeshGtidSet {
private final Map<String, UUIDSet> uuidSetsByServerId = new TreeMap<>();
// sorts on keys
public static Pattern GTID_DELIMITER = Pattern.compile(":");
- protected GtidSet(Map<String, UUIDSet> uuidSetsByServerId) {
+ protected EventMeshGtidSet(Map<String, UUIDSet> uuidSetsByServerId) {
this.uuidSetsByServerId.putAll(uuidSetsByServerId);
}
/**
* @param gtids the string representation of the GTIDs.
*/
- public GtidSet(String gtids) {
+ public EventMeshGtidSet(String gtids) {
new
com.github.shyiko.mysql.binlog.GtidSet(gtids).getUUIDSets().forEach(uuidSet -> {
uuidSetsByServerId.put(uuidSet.getServerId().toString(), new
UUIDSet(uuidSet));
});
@@ -58,12 +58,12 @@ public final class GtidSet {
}
/**
- * Obtain a copy of this {@link GtidSet} except with only the GTID ranges
that have server UUIDs that match the given predicate.
+ * Obtain a copy of this {@link EventMeshGtidSet} except with only the
GTID ranges that have server UUIDs that match the given predicate.
*
* @param sourceFilter the predicate that returns whether a server UUID is
to be included
* @return the new GtidSet, or this object if {@code sourceFilter} is
null; never null
*/
- public GtidSet retainAll(Predicate<String> sourceFilter) {
+ public EventMeshGtidSet retainAll(Predicate<String> sourceFilter) {
if (sourceFilter == null) {
return this;
}
@@ -71,7 +71,7 @@ public final class GtidSet {
.stream()
.filter(entry -> sourceFilter.test(entry.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- return new GtidSet(newSets);
+ return new EventMeshGtidSet(newSets);
}
/**
@@ -99,7 +99,7 @@ public final class GtidSet {
* @param other the other set of GTIDs; may be null
* @return {@code true} if all of the GTIDs in this set are completely
contained within the supplied set of GTIDs, or {@code false} otherwise
*/
- public boolean isContainedWithin(GtidSet other) {
+ public boolean isContainedWithin(EventMeshGtidSet other) {
if (other == null) {
return false;
}
@@ -116,19 +116,19 @@ public final class GtidSet {
}
/**
- * Obtain a copy of this {@link GtidSet} except overwritten with all of
the GTID ranges in the supplied {@link GtidSet}.
+ * Obtain a copy of this {@link EventMeshGtidSet} except overwritten with
all of the GTID ranges in the supplied {@link EventMeshGtidSet}.
*
- * @param other the other {@link GtidSet} with ranges to add/overwrite on
top of those in this set;
+ * @param other the other {@link EventMeshGtidSet} with ranges to
add/overwrite on top of those in this set;
* @return the new GtidSet, or this object if {@code other} is null or
empty; never null
*/
- public GtidSet with(GtidSet other) {
+ public EventMeshGtidSet with(EventMeshGtidSet other) {
if (other == null || other.uuidSetsByServerId.isEmpty()) {
return this;
}
Map<String, UUIDSet> newSet = new HashMap<>();
newSet.putAll(this.uuidSetsByServerId);
newSet.putAll(other.uuidSetsByServerId);
- return new GtidSet(newSet);
+ return new EventMeshGtidSet(newSet);
}
/**
@@ -136,7 +136,7 @@ public final class GtidSet {
*
* @return The starting state of the GTID set.
*/
- public GtidSet getGtidSetBeginning() {
+ public EventMeshGtidSet getGtidSetBeginning() {
// Create a new map to store the GTID set
Map<String, UUIDSet> newSet = new HashMap<>();
@@ -147,7 +147,7 @@ public final class GtidSet {
}
// Create and return a new GTID set object using the new map
- return new GtidSet(newSet);
+ return new EventMeshGtidSet(newSet);
}
public boolean contains(String gtid) {
@@ -161,7 +161,7 @@ public final class GtidSet {
return uuidSet.contains(transactionId);
}
- public GtidSet subtract(GtidSet other) {
+ public EventMeshGtidSet subtract(EventMeshGtidSet other) {
if (other == null) {
return this;
}
@@ -170,7 +170,7 @@ public final class GtidSet {
.filter(entry ->
!entry.getValue().isContainedWithin(other.forServerWithId(entry.getKey())))
.map(entry -> new AbstractMap.SimpleEntry<>(entry.getKey(),
entry.getValue().subtract(other.forServerWithId(entry.getKey()))))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- return new GtidSet(newSets);
+ return new EventMeshGtidSet(newSets);
}
@Override
@@ -183,8 +183,8 @@ public final class GtidSet {
if (obj == this) {
return true;
}
- if (obj instanceof GtidSet) {
- GtidSet that = (GtidSet) obj;
+ if (obj instanceof EventMeshGtidSet) {
+ EventMeshGtidSet that = (EventMeshGtidSet) obj;
return this.uuidSetsByServerId.equals(that.uuidSetsByServerId);
}
return false;
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlConstants.java
similarity index 64%
copy from
eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
copy to
eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlConstants.java
index f70819212..44f10cee6 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlConstants.java
@@ -15,27 +15,20 @@
* limitations under the License.
*/
-package org.apache.eventmesh.connector.jdbc;
+package org.apache.eventmesh.connector.jdbc.source.dialect.mysql;
-public final class JdbcConnectData {
+import java.util.HashSet;
+import java.util.Set;
- private Payload payload = new Payload();
+public final class MysqlConstants {
- private Schema schema;
+ public static final Set<String> DEFAULT_EXCLUDE_DATABASE = new HashSet<>();
- public Payload getPayload() {
- return payload;
+ static {
+ DEFAULT_EXCLUDE_DATABASE.add("information_schema");
+ DEFAULT_EXCLUDE_DATABASE.add("mysql");
+ DEFAULT_EXCLUDE_DATABASE.add("performance_schema");
+ DEFAULT_EXCLUDE_DATABASE.add("sys");
}
- public void setPayload(Payload payload) {
- this.payload = payload;
- }
-
- public Schema getSchema() {
- return schema;
- }
-
- public void setSchema(Schema schema) {
- this.schema = schema;
- }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlJdbcContext.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlJdbcContext.java
index 8dc64d041..c67e065c2 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlJdbcContext.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlJdbcContext.java
@@ -36,12 +36,19 @@ public class MysqlJdbcContext extends
UniversalJdbcContext<MysqlPartition, Mysql
private volatile long currentHandleEventSize = 0;
- private volatile boolean isStartTransaction = false;
+ private volatile boolean onTransaction = false;
+ //need to load from store when start
private String restartGtidSet;
private String currentGtidSet;
+ private String restartBinlogFilename;
+
+ private long restartBinlogPosition;
+
+ private String transactionId;
+
private JdbcSourceConfig jdbcSourceConfig;
public MysqlJdbcContext(PartitionOffSetContextPair<MysqlPartition,
MysqlOffsetContext> poCtx, MysqlAntlr4DdlParser parser) {
@@ -80,19 +87,22 @@ public class MysqlJdbcContext extends
UniversalJdbcContext<MysqlPartition, Mysql
}
public void startTransaction() {
- this.isStartTransaction = true;
+ this.onTransaction = true;
}
public void commitTransaction() {
- this.isStartTransaction = false;
+ this.onTransaction = false;
+ this.restartGtidSet = this.currentGtidSet;
+ this.restartBinlogFilename = sourceInfo.getCurrentBinlogFileName();
+ this.restartBinlogPosition = sourceInfo.getCurrentBinlogPosition() +
this.currentHandleEventSize;
+ resetTransactionId();
}
public void complete() {
- //TODO: What to do?
this.currentHandleEventSize = 0;
}
- public void setCompletedGtidSet(String gtidSet) {
+ public void completedGtidSet(String gtidSet) {
if (StringUtils.isNotBlank(gtidSet)) {
String trimmedGtidSet = gtidSet.replace("\n", "").replace("\r",
"");
this.currentGtidSet = trimmedGtidSet;
@@ -101,8 +111,14 @@ public class MysqlJdbcContext extends
UniversalJdbcContext<MysqlPartition, Mysql
}
public String getGtidSet() {
- return this.currentGtidSet != null ? this.currentGtidSet : null;
+ return this.currentGtidSet;
}
+ public void beginGtid(String gtid) {
+ this.sourceInfo.beginGtid(gtid);
+ }
+ private void resetTransactionId() {
+ transactionId = null;
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlSourceInfo.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlSourceInfo.java
index f362c9be9..7ff536aaa 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlSourceInfo.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/mysql/MysqlSourceInfo.java
@@ -27,10 +27,13 @@ public class MysqlSourceInfo implements SourceInfo {
private String currentBinlogFileName;
- private long currentBinlogPosition = 0L;
+ private volatile long currentBinlogPosition = 0L;
private int currentRowNumber = 0;
+ // Support mysql gtid
+ private String currentGtid;
+
public void setBinlogPosition(String binlogFileName, long
beginProcessPosition) {
if (binlogFileName != null) {
this.currentBinlogFileName = binlogFileName;
@@ -39,4 +42,8 @@ public class MysqlSourceInfo implements SourceInfo {
this.currentBinlogPosition = beginProcessPosition;
this.currentRowNumber = 0;
}
+
+ public void beginGtid(String gtid) {
+ this.currentGtid = gtid;
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/AbstractSnapshotEngine.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/AbstractSnapshotEngine.java
index 348bb7efa..598068d20 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/AbstractSnapshotEngine.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/AbstractSnapshotEngine.java
@@ -18,6 +18,8 @@
package org.apache.eventmesh.connector.jdbc.source.dialect.snapshot;
import org.apache.eventmesh.common.ThreadPoolFactory;
+import org.apache.eventmesh.connector.jdbc.DataChanges;
+import org.apache.eventmesh.connector.jdbc.DataChanges.Builder;
import org.apache.eventmesh.connector.jdbc.DatabaseDialect;
import org.apache.eventmesh.connector.jdbc.Field;
import org.apache.eventmesh.connector.jdbc.JdbcContext;
@@ -217,6 +219,21 @@ public abstract class AbstractSnapshotEngine<DbDialect
extends DatabaseDialect<J
universalJdbcContext.withTableId(tableId);
return () -> {
JdbcConnection connection = connectionPool.poll();
+ MysqlSourceMateData sourceMateData =
MysqlSourceMateData.newBuilder()
+ .name(sourceConnectorConfig.getName())
+ .snapshot(true)
+ .withTableId(tableId)
+ .serverId(sourceConnectorConfig.getMysqlConfig().getServerId())
+ .build();
+ TableSchema tableSchema =
universalJdbcContext.getCatalogTableSet().getTableSchema(tableId);
+ Field field = new
Field().withField("after").withType("field").withName("payload.after").withRequired(false);
+ List<? extends Column> columns = tableSchema.getColumns();
+ if (CollectionUtils.isNotEmpty(columns)) {
+ List<Field> fields = columns.stream()
+ .map(col -> new Field(col.getDataType().getName(),
col.isNotNull(), col.getName(), tableId.toString()))
+ .collect(Collectors.toList());
+ field.withRequired(true).withFields(fields);
+ }
try (Statement statement =
connection.createStatement(jdbcSourceConfig.getSourceConnectorConfig().getSnapshotFetchSize(),
100)) {
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()) {
@@ -227,23 +244,10 @@ public abstract class AbstractSnapshotEngine<DbDialect
extends DatabaseDialect<J
for (int index = 1; index <= columnCount; ++index) {
values.put(resultSet.getMetaData().getColumnName(index),
resultSet.getObject(index));
}
- payload.withAfterValue(values);
- MysqlSourceMateData sourceMateData =
MysqlSourceMateData.newBuilder()
- .name(sourceConnectorConfig.getName())
- .snapshot(true)
- .withTableId(tableId)
-
.serverId(sourceConnectorConfig.getMysqlConfig().getServerId())
- .build();
+ Builder builder = DataChanges.newBuilder();
+ builder.withAfter(values);
+ payload.withDataChanges(builder.build());
payload.withSource(sourceMateData);
- TableSchema tableSchema =
universalJdbcContext.getCatalogTableSet().getTableSchema(tableId);
- Field field = new
Field().withField("after").withType("field").withName("payload.after").withRequired(false);
- List<? extends Column> columns = tableSchema.getColumns();
- if (CollectionUtils.isNotEmpty(columns)) {
- List<Field> fields = columns.stream()
- .map(col -> new Field(col.getDataType().getName(),
col.isNotNull(), col.getName(), tableId.toString()))
- .collect(Collectors.toList());
- field.withRequired(true).withFields(fields);
- }
event.getJdbcConnectData().setSchema(new
Schema(Collections.singletonList(field)));
eventQueue.put(event);
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/mysql/MysqlSnapshotEngine.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/mysql/MysqlSnapshotEngine.java
index 77f054bed..cc8e6eafc 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/mysql/MysqlSnapshotEngine.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/mysql/MysqlSnapshotEngine.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.connector.jdbc.event.Event;
import org.apache.eventmesh.connector.jdbc.event.EventConsumer;
import org.apache.eventmesh.connector.jdbc.source.config.JdbcSourceConfig;
import org.apache.eventmesh.connector.jdbc.source.config.MysqlConfig;
+import org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlConstants;
import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlDatabaseDialect;
import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlDialectSql;
import
org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlJdbcContext;
@@ -38,7 +39,6 @@ import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.OptionalLong;
@@ -52,15 +52,6 @@ import lombok.extern.slf4j.Slf4j;
public class MysqlSnapshotEngine extends
AbstractSnapshotEngine<MysqlDatabaseDialect, MysqlJdbcContext,
MysqlPartition, MysqlOffsetContext, MysqlJdbcConnection> {
- private static final Set<String> DEFAULT_EXCLUDE_DATABASE = new
HashSet<>();
-
- static {
- DEFAULT_EXCLUDE_DATABASE.add("information_schema");
- DEFAULT_EXCLUDE_DATABASE.add("mysql");
- DEFAULT_EXCLUDE_DATABASE.add("performance_schema");
- DEFAULT_EXCLUDE_DATABASE.add("sys");
- }
-
private volatile boolean globalLockAcquired = false;
private volatile boolean tableLockAcquired = false;
@@ -77,7 +68,7 @@ public class MysqlSnapshotEngine extends
@Override
protected Set<String> defaultExcludeDatabase() {
- return DEFAULT_EXCLUDE_DATABASE;
+ return MysqlConstants.DEFAULT_EXCLUDE_DATABASE;
}
@Override
@@ -133,7 +124,7 @@ public class MysqlSnapshotEngine extends
jdbcContext.setBinlogStartPoint(binlogFilename, position);
if (resultSet.getMetaData().getColumnCount() >= 5) {
final String gtidSet = resultSet.getString(5);
- jdbcContext.setCompletedGtidSet(gtidSet);
+ jdbcContext.completedGtidSet(gtidSet);
log.info("Using binlog '{}' at position '{}' and gtid
'{}'", binlogFilename, position, gtidSet);
} else {
log.info("Using binlog '{}' at position '{}' ",
binlogFilename, position);
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/AbstractColumnEditorImpl.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/AbstractColumnEditorImpl.java
index 32c416c5b..4f7d1d5b4 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/AbstractColumnEditorImpl.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/AbstractColumnEditorImpl.java
@@ -66,6 +66,8 @@ public abstract class AbstractColumnEditorImpl<CE extends
ColumnEditor, Col exte
private boolean optional;
+ private int order;
+
public AbstractColumnEditorImpl(String name) {
this.name = name;
}
@@ -136,6 +138,19 @@ public abstract class AbstractColumnEditorImpl<CE extends
ColumnEditor, Col exte
return (CE) this;
}
+ /**
+ * Sets the order or position of the column within a table.
+ *
+ * @param order The order or position of the column.
+ * @return The column editor instance.
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ public CE withOrder(int order) {
+ this.order = order;
+ return (CE) this;
+ }
+
/**
* Sets the length of the column (if applicable).
*
@@ -266,4 +281,8 @@ public abstract class AbstractColumnEditorImpl<CE extends
ColumnEditor, Col exte
public boolean isOptional() {
return optional;
}
+
+ public int ofOrder() {
+ return this.order;
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/ColumnEditor.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/ColumnEditor.java
index c72c3f22e..30a039b63 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/ColumnEditor.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/ColumnEditor.java
@@ -69,6 +69,14 @@ public interface ColumnEditor<CE extends ColumnEditor, Col
extends Column> {
*/
CE withEventMeshType(EventMeshDataType<?> eventMeshType);
+ /**
+ * Sets the order or position of the column within a table.
+ *
+ * @param order The order or position of the column.
+ * @return The column editor instance.
+ */
+ CE withOrder(int order);
+
/**
* Sets the length of the column (if applicable).
*
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableId.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableId.java
index 2e6db21d3..37be43329 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableId.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableId.java
@@ -159,4 +159,8 @@ public class TableId implements Serializable {
return tableId(tableId.getCatalogName(), tableId.getSchemaName(),
tableId.getTableName());
}
}
+
+ public TableId copy() {
+ return new TableId(this.catalogName, this.schemaName, this.tableName);
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableSchema.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableSchema.java
index 94f88052e..3eadd78d5 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableSchema.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableSchema.java
@@ -28,12 +28,10 @@ import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;
-import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
-@AllArgsConstructor
@NoArgsConstructor
public class TableSchema implements Serializable {
@@ -49,6 +47,8 @@ public class TableSchema implements Serializable {
*/
private List<? extends Column> columns;
+ private Map<Integer, ? extends Column> orderColumnMap;
+
/**
* The primary key of the table.
*/
@@ -58,6 +58,17 @@ public class TableSchema implements Serializable {
private String comment;
+ public TableSchema(TableId tableId, Map<String, ? extends Column>
columnMap, List<? extends Column> columns,
+ Map<Integer, ? extends Column> orderColumnMap, PrimaryKey primaryKey,
List<UniqueKey> uniqueKeys, String comment) {
+ this.tableId = tableId;
+ this.columnMap = columnMap;
+ this.columns = columns;
+ this.orderColumnMap = orderColumnMap;
+ this.primaryKey = primaryKey;
+ this.uniqueKeys = uniqueKeys;
+ this.comment = comment;
+ }
+
public TableSchema(String name) {
this.tableId.setTableName(name);
}
@@ -78,6 +89,7 @@ public class TableSchema implements Serializable {
private TableId tableId;
private Map<String, Column> columnMap;
+ private Map<Integer, Column> orderColumnMap;
private List<Column> columns;
private PrimaryKey primaryKey;
private List<UniqueKey> uniqueKeys;
@@ -118,7 +130,7 @@ public class TableSchema implements Serializable {
}
public TableSchema build() {
- return new TableSchema(tableId, columnMap, columns, primaryKey,
uniqueKeys, comment);
+ return new TableSchema(tableId, columnMap, columns,
orderColumnMap, primaryKey, uniqueKeys, comment);
}
}
@@ -131,6 +143,7 @@ public class TableSchema implements Serializable {
private TableId tableId = new TableId();
private Map<String, MysqlColumn> columnMap;
+ private Map<Integer, MysqlColumn> orderColumnMap;
private List<MysqlColumn> columns;
private PrimaryKey primaryKey;
private List<UniqueKey> uniqueKeys;
@@ -155,6 +168,8 @@ public class TableSchema implements Serializable {
this.columns = columns;
this.columnMap = Optional.ofNullable(columns).orElse(new
ArrayList<>(0)).stream()
.collect(Collectors.toMap(MysqlColumn::getName,
Function.identity()));
+ this.orderColumnMap = Optional.ofNullable(columns).orElse(new
ArrayList<>(0)).stream()
+ .collect(Collectors.toMap(MysqlColumn::getOrder,
Function.identity()));
return this;
}
@@ -184,7 +199,7 @@ public class TableSchema implements Serializable {
}
public MysqlTableSchema build() {
- return new MysqlTableSchema(tableId, columnMap, columns,
primaryKey, uniqueKeys, comment);
+ return new MysqlTableSchema(tableId, columnMap, columns,
orderColumnMap, primaryKey, uniqueKeys, comment, tableOptions);
}
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/mysql/MysqlColumnEditorImpl.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/mysql/MysqlColumnEditorImpl.java
index 2b7db0591..c7b7648af 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/mysql/MysqlColumnEditorImpl.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/mysql/MysqlColumnEditorImpl.java
@@ -78,6 +78,6 @@ public class MysqlColumnEditorImpl extends
AbstractColumnEditorImpl<MysqlColumnE
@Override
public MysqlColumn build() {
return MysqlColumn.of(ofName(), ofEventMeshDataType(), ofJdbcType(),
ofColumnLength(), ofScale(), isNotNull(), ofComment(), ofDefaultValue(),
- ofDefaultValueExpression(), autoIncremented, generated,
collationName);
+ ofDefaultValueExpression(), autoIncremented, generated,
collationName, ofOrder());
}
}
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/mysql/MysqlTableSchema.java
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/mysql/MysqlTableSchema.java
index 8e61d7af9..c26ca5c3c 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/mysql/MysqlTableSchema.java
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/mysql/MysqlTableSchema.java
@@ -30,9 +30,10 @@ public class MysqlTableSchema extends TableSchema {
private Options tableOptions;
- public MysqlTableSchema(TableId tableId, Map<String, MysqlColumn>
columnMap, List<MysqlColumn> columns, PrimaryKey primaryKey,
- List<UniqueKey> uniqueKeys, String comment) {
- super(tableId, columnMap, columns, primaryKey, uniqueKeys, comment);
+ public MysqlTableSchema(TableId tableId, Map<String, MysqlColumn>
columnMap, List<MysqlColumn> columns, Map<Integer, MysqlColumn> orderColumnMap,
+ PrimaryKey primaryKey, List<UniqueKey> uniqueKeys, String comment,
Options tableOptions) {
+ super(tableId, columnMap, columns, orderColumnMap, primaryKey,
uniqueKeys, comment);
+ this.tableOptions = tableOptions;
}
public MysqlTableSchema() {
diff --git
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngineFactory
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngineFactory
index 8b07afd30..06eb1b99e 100644
---
a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngineFactory
+++
b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngineFactory
@@ -13,4 +13,4 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-#mysql=org.apache.eventmesh.connector.jdbc.source.dialect.cdc.mysql.MysqlCdcEngineFactory
+mysql=org.apache.eventmesh.connector.jdbc.source.dialect.cdc.mysql.MysqlCdcEngineFactory
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
