You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by ch...@apache.org on 2023/08/29 08:02:31 UTC
[eventmesh] branch master updated: Add connector jdbc interface (#4332)
This is an automated email from the ASF dual-hosted git repository.
chenguangsheng 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 eea1d01af Add connector jdbc interface (#4332)
eea1d01af is described below
commit eea1d01afab6c73e66cd23ac6c5a9f1858d03a00
Author: mxsm <lj...@gmail.com>
AuthorDate: Tue Aug 29 16:02:20 2023 +0800
Add connector jdbc interface (#4332)
---
.gitignore | 3 +-
build.gradle | 5 +
.../eventmesh/common/ResetCountDownLatch.java | 180 +++++++
.../org/apache/eventmesh/common/ThreadWrapper.java | 130 +++++
.../eventmesh/common/ResetCountDownLatchTest.java | 79 +++
.../apache/eventmesh/common/ThreadWrapperTest.java | 134 +++++
.../eventmesh-connector-jdbc/build.gradle | 53 ++
.../eventmesh/connector/jdbc/CatalogChanges.java | 140 +++++
.../connector/jdbc/DataTypeConvertor.java | 78 +++
.../eventmesh/connector/jdbc/DatabaseDialect.java | 74 +++
.../org/apache/eventmesh/connector/jdbc/Field.java | 31 +-
.../eventmesh/connector/jdbc/JdbcConnectData.java | 41 +-
.../eventmesh/connector/jdbc/JdbcContext.java | 27 +-
.../connector/jdbc/JdbcDriverMetaData.java | 49 +-
.../eventmesh/connector/jdbc/OffsetContext.java | 43 +-
.../apache/eventmesh/connector/jdbc/Partition.java | 35 +-
.../apache/eventmesh/connector/jdbc/Payload.java | 72 +++
.../apache/eventmesh/connector/jdbc/Schema.java | 28 +-
.../connector/jdbc/config/JdbcConfig.java | 63 +++
.../connector/jdbc/config/JdbcServerConfig.java | 36 +-
.../connector/jdbc/connection/JdbcConnection.java | 596 +++++++++++++++++++++
.../jdbc/connection/JdbcConnectionProvider.java | 50 ++
.../jdbc/connection/PreparedParameter.java | 32 +-
.../connector/jdbc/ddl/AbstractDdlParser.java | 60 +++
.../eventmesh/connector/jdbc/ddl/DdlParser.java | 40 +-
.../connector/jdbc/ddl/DdlParserCallback.java | 36 +-
.../connector/jdbc/event/DataChangeEvent.java | 34 +-
.../connector/jdbc/event/DataChangeEventType.java | 29 +-
.../eventmesh/connector/jdbc/event/Event.java | 36 +-
.../connector/jdbc/event/EventConsumer.java | 34 +-
.../connector/jdbc/event/EventHandler.java | 35 +-
.../connector/jdbc/event/SchemaChangeEvent.java | 33 +-
.../jdbc/event/SchemaChangeEventType.java | 38 +-
.../connector/jdbc/exception/CatalogException.java | 80 +++
.../jdbc/exception/DataTypeConvertException.java | 82 +++
.../jdbc/exception/DatabaseNotExistException.java | 80 +++
.../jdbc/exception/JdbcConnectionException.java | 28 +-
.../jdbc/exception/TableNotExistException.java | 79 +++
.../connector/jdbc/server/JdbcConnectorServer.java | 37 +-
.../source/AbstractEventMeshJdbcEventTask.java | 44 +-
.../jdbc/source/AbstractJdbcTaskManager.java | 67 +++
.../eventmesh/connector/jdbc/source/Engine.java | 43 +-
.../connector/jdbc/source/EventDispatcher.java | 50 ++
.../jdbc/source/EventMeshJdbcEventTask.java | 45 ++
.../connector/jdbc/source/EventMeshJdbcTask.java | 35 +-
.../jdbc/source/JdbcAllFactoryLoader.java | 72 +++
.../connector/jdbc/source/JdbcSourceConnector.java | 207 +++++++
.../connector/jdbc/source/JdbcTaskManager.java | 49 ++
.../jdbc/source/SourceEventMeshJdbcEventTask.java | 77 +++
.../jdbc/source/SourceJdbcTaskManager.java | 96 ++++
.../connector/jdbc/source/SourceMateData.java | 64 +++
.../jdbc/source/TaskManagerCoordinator.java | 116 ++++
.../connector/jdbc/source/TaskManagerListener.java | 38 +-
.../jdbc/source/config/JdbcSourceConfig.java | 32 +-
.../connector/jdbc/source/config/MysqlConfig.java | 56 +-
.../jdbc/source/config/SourceConnectorConfig.java | 78 +++
.../source/dialect/DatabaseDialectFactory.java | 39 +-
.../jdbc/source/dialect/cdc/CdcEngine.java | 50 ++
.../jdbc/source/dialect/cdc/CdcEngineFactory.java | 49 ++
.../dialect/cdc/RandomTaskSelectStrategy.java | 46 ++
.../source/dialect/cdc/TaskSelectStrategy.java | 36 +-
.../source/dialect/snapshot/SnapshotEngine.java | 45 ++
.../dialect/snapshot/SnapshotEngineFactory.java | 43 ++
.../source/dialect/snapshot/SnapshotResult.java | 43 +-
.../jdbc/source/dialect/snapshot/SnapshotType.java | 35 +-
.../connector/jdbc/table/catalog/Catalog.java | 96 ++++
.../jdbc/table/catalog/CatalogSchema.java | 43 +-
.../connector/jdbc/table/catalog/CatalogTable.java | 45 +-
.../connector/jdbc/table/catalog/Column.java | 102 ++++
.../connector/jdbc/table/catalog/ColumnEditor.java | 149 ++++++
.../jdbc/table/catalog/DefaultColumn.java | 51 ++
.../table/catalog/DefaultColumnEditorImpl.java | 258 +++++++++
.../connector/jdbc/table/catalog/Options.java | 28 +-
.../connector/jdbc/table/catalog/PrimaryKey.java | 59 ++
.../connector/jdbc/table/catalog/Table.java | 38 +-
.../connector/jdbc/table/catalog/TableId.java | 156 ++++++
.../connector/jdbc/table/catalog/TableSchema.java | 112 ++++
.../connector/jdbc/table/catalog/UniqueKey.java | 56 +-
.../jdbc/table/type/EventMeshDataType.java | 38 +-
.../eventmesh/connector/jdbc/table/type/Pair.java | 35 +-
.../connector/jdbc/table/type/RowHandleMode.java | 86 +++
.../connector/jdbc/table/type/SQLType.java | 96 ++++
.../connector/jdbc/utils/JdbcStringUtils.java | 50 ++
...ctor.jdbc.source.dialect.DatabaseDialectFactory | 16 +
...nector.jdbc.source.dialect.cdc.CdcEngineFactory | 16 +
...c.source.dialect.snapshot.SnapshotEngineFactory | 16 +
.../src/main/resources/log4j2.xml | 31 ++
.../src/main/resources/server-config.yml | 19 +
.../src/main/resources/source-config.yml | 60 +++
.../connector/jdbc/utils/JdbcStringUtilsTest.java | 55 ++
.../eventmesh/spi/EventMeshExtensionType.java | 4 +-
settings.gradle | 1 +
92 files changed, 5159 insertions(+), 852 deletions(-)
diff --git a/.gitignore b/.gitignore
index e75f1a127..95bfd3fd5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -47,4 +47,5 @@ bld/
[Bb]in/
[Oo]bj/
[Ll]og/
-[Ll]ogs/
\ No newline at end of file
+[Ll]ogs/
+**/org/apache/eventmesh/connector/jdbc/antlr4/autogeneration/*
diff --git a/build.gradle b/build.gradle
index e368eb58c..8f2293ae0 100644
--- a/build.gradle
+++ b/build.gradle
@@ -90,6 +90,7 @@ allprojects {
.exclude('**/org/apache/eventmesh/common/protocol/grpc/cloudevents**')
.exclude('**/org/apache/eventmesh/connector/openfunction/client/EventMeshGrpcService**')
.exclude('**/org/apache/eventmesh/connector/openfunction/client/CallbackServiceGrpc**')
+ .exclude('**/org/apache/eventmesh/connector/jdbc/antlr**')
dependencies {
repositories {
@@ -249,6 +250,9 @@ subprojects {
rulesMinimumPriority = 5
ruleSets = ["category/java/errorprone.xml", "category/java/bestpractices.xml"]
ignoreFailures = true
+ pmdMain {
+ excludes = ["**/org/apache/eventmesh/connector/jdbc/antlr4/autogeneration/**"]
+ }
}
jar {
@@ -341,6 +345,7 @@ subprojects {
javadoc {
source = sourceSets.main.java
destinationDir = reporting.file("javadoc")
+ options.encoding = "UTF-8"
}
task packageJavadoc(type: Jar, dependsOn: ['javadoc']) {
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/ResetCountDownLatch.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/ResetCountDownLatch.java
new file mode 100644
index 000000000..c04cc6cde
--- /dev/null
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/ResetCountDownLatch.java
@@ -0,0 +1,180 @@
+/*
+ * 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;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.AbstractQueuedSynchronizer;
+
+/**
+ * ResetCountDownLatch can reset
+ *
+ * @see java.util.concurrent.CountDownLatch
+ */
+public class ResetCountDownLatch {
+
+ private final RestSync restSync;
+
+ public ResetCountDownLatch(int count) {
+ this.restSync = new RestSync(count);
+ }
+
+
+ /**
+ * Causes the current thread to wait until the latch has counted down to zero, unless the thread is {@linkplain Thread#interrupt interrupted}.
+ *
+ * <p>If the current count is zero then this method returns immediately.
+ *
+ * <p>If the current count is greater than zero then the current
+ * thread becomes disabled for thread scheduling purposes and lies dormant until one of two things happen:
+ * <ul>
+ * <li>The count reaches zero due to invocations of the
+ * {@link #countDown} method; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread.
+ * </ul>
+ *
+ * <p>If the current thread:
+ * <ul>
+ * <li>has its interrupted status set on entry to this method; or
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting,
+ * </ul>
+ * then {@link InterruptedException} is thrown and the current thread's
+ * interrupted status is cleared.
+ *
+ * @throws InterruptedException if the current thread is interrupted while waiting
+ */
+ public void await() throws InterruptedException {
+ restSync.acquireSharedInterruptibly(1);
+ }
+
+ /**
+ * Causes the current thread to wait until the latch has counted down to zero, unless the thread is {@linkplain Thread#interrupt interrupted}, or
+ * the specified waiting time elapses.
+ *
+ * <p>If the current count is zero then this method returns immediately
+ * with the value {@code true}.
+ *
+ * <p>If the current count is greater than zero then the current
+ * thread becomes disabled for thread scheduling purposes and lies dormant until one of three things happen:
+ * <ul>
+ * <li>The count reaches zero due to invocations of the
+ * {@link #countDown} method; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread; or
+ * <li>The specified waiting time elapses.
+ * </ul>
+ *
+ * <p>If the count reaches zero then the method returns with the
+ * value {@code true}.
+ *
+ * <p>If the current thread:
+ * <ul>
+ * <li>has its interrupted status set on entry to this method; or
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting,
+ * </ul>
+ * then {@link InterruptedException} is thrown and the current thread's
+ * interrupted status is cleared.
+ *
+ * <p>If the specified waiting time elapses then the value {@code false}
+ * is returned. If the time is less than or equal to zero, the method
+ * will not wait at all.
+ *
+ * @param timeout the maximum time to wait
+ * @param unit the time unit of the {@code timeout} argument
+ * @return {@code true} if the count reached zero and {@code false} if the waiting time elapsed before the count reached zero
+ * @throws InterruptedException if the current thread is interrupted while waiting
+ */
+ public boolean await(long timeout, TimeUnit unit)
+ throws InterruptedException {
+ return restSync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
+ }
+
+
+ /**
+ * Decrements the count of the latch, releasing all waiting threads if the count reaches zero.
+ *
+ * <p>If the current count is greater than zero then it is decremented.
+ * If the new count is zero then all waiting threads are re-enabled for thread scheduling purposes.
+ *
+ * <p>If the current count equals zero then nothing happens.
+ */
+ public void countDown() {
+ restSync.releaseShared(1);
+ }
+
+ /**
+ * Returns the current count.
+ *
+ * <p>This method is typically used for debugging and testing purposes.
+ *
+ * @return the current count
+ */
+ public int getCount() {
+ return restSync.getCount();
+ }
+
+ /**
+ * Reset the CountDownLatch
+ */
+ public void reset() {
+ restSync.reset();
+ }
+
+ /**
+ * Synchronization control For ResetCountDownLatch. Uses AQS state to represent count.
+ */
+ private static final class RestSync extends AbstractQueuedSynchronizer {
+
+ private final int initCount;
+
+ RestSync(int count) {
+ if (count < 0) {
+ throw new IllegalArgumentException("count must be greater than or equal to 0");
+ }
+ this.initCount = count;
+ setState(count);
+ }
+
+ protected void reset() {
+ setState(initCount);
+ }
+
+ int getCount() {
+ return getState();
+ }
+
+ @Override
+ protected int tryAcquireShared(int acquires) {
+ return (getState() == 0) ? 1 : -1;
+ }
+
+ @Override
+ protected boolean tryReleaseShared(int releases) {
+ for (; ; ) {
+ int count = getState();
+ if (count == 0) {
+ return false;
+ }
+ int nextCount = count - 1;
+ if (compareAndSetState(count, nextCount)) {
+ return nextCount == 0;
+ }
+ }
+ }
+ }
+}
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/ThreadWrapper.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/ThreadWrapper.java
new file mode 100644
index 000000000..0b67e69e9
--- /dev/null
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/ThreadWrapper.java
@@ -0,0 +1,130 @@
+/*
+ * 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;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public abstract class ThreadWrapper implements Runnable {
+
+ private final AtomicBoolean started = new AtomicBoolean(false);
+ protected Thread thread;
+ protected final ResetCountDownLatch waiter = new ResetCountDownLatch(1);
+ protected volatile AtomicBoolean hasWakeup = new AtomicBoolean(false);
+ protected boolean isDaemon = false;
+ protected volatile boolean isRunning = false;
+
+ public ThreadWrapper() {
+
+ }
+
+ public abstract String getThreadName();
+
+ public void start() {
+
+ if (!started.compareAndSet(false, true)) {
+ log.warn("Start thread:{} fail", getThreadName());
+ return;
+ }
+ this.thread = new Thread(this, getThreadName());
+ this.thread.setDaemon(isDaemon);
+ this.thread.start();
+ this.isRunning = true;
+ log.info("Start thread:{} success", getThreadName());
+ }
+
+ public void await() {
+ if (hasWakeup.compareAndSet(true, false)) {
+ return;
+ }
+ //reset count
+ waiter.reset();
+ try {
+ waiter.await();
+ } catch (InterruptedException e) {
+ log.error("Thread[{}] Interrupted", getThreadName(), e);
+ } finally {
+ hasWakeup.set(false);
+ }
+ }
+
+ public void await(long timeout) {
+ await(timeout, TimeUnit.MILLISECONDS);
+ }
+
+ public void await(long timeout, TimeUnit timeUnit) {
+ if (hasWakeup.compareAndSet(true, false)) {
+ return;
+ }
+ //reset count
+ waiter.reset();
+ try {
+ waiter.await(timeout, timeUnit == null ? TimeUnit.MILLISECONDS : timeUnit);
+ } catch (InterruptedException e) {
+ log.error("Thread[{}] Interrupted", getThreadName(), e);
+ } finally {
+ hasWakeup.set(false);
+ }
+ }
+
+ public void wakeup() {
+ if (hasWakeup.compareAndSet(false, true)) {
+ waiter.countDown();
+ }
+ }
+
+ public void shutdownImmediately() {
+ shutdown(true);
+ }
+
+ public void shutdown() {
+ shutdown(false);
+ }
+
+ private void shutdown(final boolean interruptThread) {
+ if (!started.compareAndSet(true, false)) {
+ return;
+ }
+ this.isRunning = false;
+ //wakeup the thread to run
+ wakeup();
+
+ try {
+ if (interruptThread) {
+ this.thread.interrupt();
+ }
+ if (!this.isDaemon) {
+ //wait main thread to wait this thread finish
+ this.thread.join(TimeUnit.SECONDS.toMillis(60));
+ }
+ } catch (InterruptedException e) {
+ log.error("Thread[{}] Interrupted", getThreadName(), e);
+ }
+ }
+
+ public void setDaemon(boolean daemon) {
+ isDaemon = daemon;
+ }
+
+ public boolean isStated() {
+ return this.started.get();
+ }
+}
diff --git a/eventmesh-common/src/test/java/org/apache/eventmesh/common/ResetCountDownLatchTest.java b/eventmesh-common/src/test/java/org/apache/eventmesh/common/ResetCountDownLatchTest.java
new file mode 100644
index 000000000..abc3229a4
--- /dev/null
+++ b/eventmesh-common/src/test/java/org/apache/eventmesh/common/ResetCountDownLatchTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ResetCountDownLatchTest {
+
+ @Test
+ public void testConstructorParameterError() {
+ try {
+ new ResetCountDownLatch(-1);
+ } catch (IllegalArgumentException e) {
+ Assert.assertEquals(e.getMessage(), "count must be greater than or equal to 0");
+ }
+ ResetCountDownLatch resetCountDownLatch = new ResetCountDownLatch(1);
+ Assert.assertEquals(1, resetCountDownLatch.getCount());
+ }
+
+ @Test
+ public void testAwaitTimeout() throws InterruptedException {
+ ResetCountDownLatch latch = new ResetCountDownLatch(1);
+ boolean await = latch.await(5, TimeUnit.MILLISECONDS);
+ Assert.assertFalse(await);
+ latch.countDown();
+ await = latch.await(5, TimeUnit.MILLISECONDS);
+ Assert.assertTrue(await);
+ }
+
+ @Test(timeout = 1000)
+ public void testCountDownAndGetCount() throws InterruptedException {
+ int count = 2;
+ ResetCountDownLatch resetCountDownLatch = new ResetCountDownLatch(count);
+ Assert.assertEquals(count, resetCountDownLatch.getCount());
+ resetCountDownLatch.countDown();
+ Assert.assertEquals(count - 1, resetCountDownLatch.getCount());
+ resetCountDownLatch.countDown();
+ resetCountDownLatch.await();
+ Assert.assertEquals(0, resetCountDownLatch.getCount());
+ }
+
+ @Test
+ public void testReset() throws InterruptedException {
+ int count = 2;
+ ResetCountDownLatch resetCountDownLatch = new ResetCountDownLatch(count);
+ resetCountDownLatch.countDown();
+ Assert.assertEquals(count - 1, resetCountDownLatch.getCount());
+ resetCountDownLatch.reset();
+ Assert.assertEquals(count, resetCountDownLatch.getCount());
+ resetCountDownLatch.countDown();
+ resetCountDownLatch.countDown();
+ resetCountDownLatch.await();
+ Assert.assertEquals(0, resetCountDownLatch.getCount());
+ resetCountDownLatch.countDown();
+ Assert.assertEquals(0, resetCountDownLatch.getCount());
+ resetCountDownLatch.reset();
+ resetCountDownLatch.countDown();
+ Assert.assertEquals(1, resetCountDownLatch.getCount());
+
+ }
+}
\ No newline at end of file
diff --git a/eventmesh-common/src/test/java/org/apache/eventmesh/common/ThreadWrapperTest.java b/eventmesh-common/src/test/java/org/apache/eventmesh/common/ThreadWrapperTest.java
new file mode 100644
index 000000000..38b0dd473
--- /dev/null
+++ b/eventmesh-common/src/test/java/org/apache/eventmesh/common/ThreadWrapperTest.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.common;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ThreadWrapperTest {
+
+ @Test
+ public void getThreadName() {
+ ThreadWrapper wrapper = createThreadWrapper(false);
+ wrapper.start();
+ Assert.assertEquals("EventMesh-Wrapper-mxsm", wrapper.thread.getName());
+ }
+
+ @Test
+ public void start() {
+ ThreadWrapper wrapper = createThreadWrapper(false);
+ wrapper.start();
+ Assert.assertTrue(wrapper.isStated());
+ }
+
+ @Test(timeout = 1000)
+ public void await() {
+ ThreadWrapper wrapper = createThreadWrapper(false);
+ wrapper.start();
+ wrapper.await(1, TimeUnit.MILLISECONDS);
+ Assert.assertFalse(wrapper.hasWakeup.get());
+ wrapper.wakeup();
+ Assert.assertTrue(wrapper.hasWakeup.get());
+ wrapper.await();
+ Assert.assertFalse(wrapper.hasWakeup.get());
+ wrapper.await(2, TimeUnit.MILLISECONDS);
+
+ }
+
+ @Test
+ public void wakeup() {
+ }
+
+ @Test
+ public void shutdown() {
+ AtomicInteger counter = new AtomicInteger();
+ ThreadWrapper wrapper = new ThreadWrapper() {
+ @Override
+ public String getThreadName() {
+ return "EventMesh-Wrapper-mxsm";
+ }
+
+ @Override
+ public void run() {
+ try {
+ TimeUnit.MILLISECONDS.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ counter.set(100);
+ }
+ };
+ wrapper.start();
+ wrapper.shutdown();
+ Assert.assertEquals(100, counter.get());
+ }
+
+ @Test
+ public void shutdownImmediately() {
+ AtomicInteger counter = new AtomicInteger();
+ ThreadWrapper wrapper = new ThreadWrapper() {
+ @Override
+ public String getThreadName() {
+ return "EventMesh-Wrapper-mxsm";
+ }
+
+ @Override
+ public void run() {
+ try {
+ TimeUnit.SECONDS.sleep(100);
+ } catch (InterruptedException e) {
+ return;
+ }
+ counter.set(100);
+ }
+ };
+ wrapper.start();
+ wrapper.shutdownImmediately();
+ Assert.assertEquals(0, counter.get());
+ }
+
+ @Test
+ public void setDaemon() {
+ ThreadWrapper threadWrapper = createThreadWrapper(true);
+ threadWrapper.start();
+ Assert.assertTrue(threadWrapper.thread.isDaemon());
+
+ ThreadWrapper threadWrapper1 = createThreadWrapper(false);
+ threadWrapper1.start();
+ Assert.assertFalse(threadWrapper1.thread.isDaemon());
+ }
+
+ private ThreadWrapper createThreadWrapper(boolean daemon) {
+ ThreadWrapper wrapper = new ThreadWrapper() {
+ @Override
+ public String getThreadName() {
+ return "EventMesh-Wrapper-mxsm";
+ }
+
+ @Override
+ public void run() {
+ // nothing to do
+ }
+ };
+ wrapper.setDaemon(daemon);
+ return wrapper;
+ }
+}
\ No newline at end of file
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/build.gradle b/eventmesh-connectors/eventmesh-connector-jdbc/build.gradle
new file mode 100644
index 000000000..e4fac1705
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/build.gradle
@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+
+plugins {
+ id 'java'
+ //id 'antlr'
+}
+
+repositories {
+ mavenCentral()
+}
+
+/*generateGrammarSource {
+ maxHeapSize = '64m'
+ arguments += ['-package', 'org.apache.eventmesh.connector.jdbc.antlr4.autogeneration', '-visitor']
+ outputDirectory = file('src/main/java/org/apache/eventmesh/connector/jdbc/antlr4/autogeneration')
+}
+
+packageSources {
+ dependsOn generateGrammarSource
+}*/
+
+dependencies {
+ //antlr("org.antlr:antlr4:4.13.0")
+ implementation 'org.antlr:antlr4-runtime:4.13.0'
+ implementation project(":eventmesh-common")
+ implementation project(":eventmesh-openconnect:eventmesh-openconnect-java")
+ implementation project(":eventmesh-spi")
+ implementation 'com.zendesk:mysql-binlog-connector-java:0.28.0'
+ implementation 'mysql:mysql-connector-java:8.0.32'
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+
+ testImplementation "org.assertj:assertj-core"
+
+ testImplementation "org.mockito:mockito-core"
+ testImplementation "org.powermock:powermock-module-junit4"
+ testImplementation "org.powermock:powermock-api-mockito2"
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/CatalogChanges.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/CatalogChanges.java
new file mode 100644
index 000000000..a58d6c305
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/CatalogChanges.java
@@ -0,0 +1,140 @@
+/*
+ * 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 org.apache.eventmesh.connector.jdbc.event.SchemaChangeEventType;
+import org.apache.eventmesh.connector.jdbc.table.catalog.CatalogSchema;
+import org.apache.eventmesh.connector.jdbc.table.catalog.Column;
+import org.apache.eventmesh.connector.jdbc.table.catalog.Table;
+
+import java.util.List;
+
+import lombok.Data;
+
+@Data
+/**
+ * Represents changes in a catalog, such as schema or table modifications.
+ */
+public class CatalogChanges {
+
+ /**
+ * The type of change (e.g., "T(Table)" or "D(Database)").
+ * <pr>
+ * {@link SchemaChangeEventType#type}
+ * </pr>
+ */
+ private String type;
+
+ /**
+ * The specific operation type (e.g., "C(Create)", "D(Drop)", "A(Alert)").
+ * <pr>
+ * {@link SchemaChangeEventType#operationType}
+ * </pr>
+ */
+ private String operationType;
+ // The catalog schema associated with the changes
+ private CatalogSchema catalog;
+ // The table associated with the changes
+ private Table table;
+ // The list of columns affected by the changes
+ private List<? extends Column> columns;
+
+ private CatalogChanges(String type, String operationType, CatalogSchema catalog, Table table,
+ List<? extends Column> columns) {
+ this.type = type;
+ this.operationType = operationType;
+ this.catalog = catalog;
+ this.table = table;
+ this.columns = columns;
+ }
+
+ /**
+ * Creates a new instance of the Builder for constructing CatalogChanges.
+ *
+ * @return The Builder instance.
+ */
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ /**
+ * Builder class for constructing CatalogChanges.
+ */
+ public static class Builder {
+
+ private String type;
+ private String operationType;
+ private CatalogSchema catalog;
+ private Table table;
+ private List<? extends Column> columns;
+
+ /**
+ * Sets the operation type for the change.
+ *
+ * @param changeEventType The SchemaChangeEventType representing the change.
+ * @return The Builder instance.
+ */
+ public Builder operationType(SchemaChangeEventType changeEventType) {
+ this.type = changeEventType.ofType();
+ this.operationType = changeEventType.ofOperationType();
+ return this;
+ }
+
+ /**
+ * Sets the catalog schema associated with the changes.
+ *
+ * @param catalog The CatalogSchema instance.
+ * @return The Builder instance.
+ */
+ public Builder catalog(CatalogSchema catalog) {
+ this.catalog = catalog;
+ return this;
+ }
+
+ /**
+ * Sets the table associated with the changes.
+ *
+ * @param table The Table instance.
+ * @return The Builder instance.
+ */
+ public Builder table(Table table) {
+ this.table = table;
+ return this;
+ }
+
+ /**
+ * Sets the list of columns affected by the changes.
+ *
+ * @param columns The list of Column instances.
+ * @return The Builder instance.
+ */
+ public Builder columns(List<? extends Column> columns) {
+ this.columns = columns;
+ return this;
+ }
+
+ /**
+ * Builds a new CatalogChanges instance.
+ *
+ * @return The constructed CatalogChanges instance.
+ */
+ public CatalogChanges build() {
+ return new CatalogChanges(type, operationType, catalog, table, columns);
+ }
+ }
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DataTypeConvertor.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DataTypeConvertor.java
new file mode 100644
index 000000000..38dee7ea5
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DataTypeConvertor.java
@@ -0,0 +1,78 @@
+/*
+ * 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 org.apache.eventmesh.connector.jdbc.exception.DataTypeConvertException;
+import org.apache.eventmesh.connector.jdbc.table.type.EventMeshDataType;
+
+import java.sql.JDBCType;
+import java.util.Map;
+
+/**
+ * convert data types between EventMesh and connector.
+ */
+public interface DataTypeConvertor<T> {
+
+ /**
+ * Converts a string representation of a connector data type to the corresponding JDBCType.
+ *
+ * @param connectorDataType The string representation of the connector data type.
+ * @return The corresponding JDBCType, or null if the connector data type is not recognized.
+ */
+ JDBCType toJDBCType(String connectorDataType);
+
+
+ /**
+ * Converts a connector data type to an EventMesh data type.
+ *
+ * @param connectorDataType The connector data type to be converted.
+ * @return The converted EventMesh data type.
+ * @throws DataTypeConvertException If the conversion fails.
+ */
+ EventMeshDataType<?> toEventMeshType(String connectorDataType) throws DataTypeConvertException;
+
+ /**
+ * Converts JDBCType and dataTypeProperties to EventMeshDataType.
+ *
+ * @param jdbcType the JDBCType to be converted
+ * @param dataTypeProperties the properties of the data type
+ * @return the converted EventMeshDataType
+ * @throws DataTypeConvertException if there is an error during conversion
+ */
+ EventMeshDataType<?> toEventMeshType(JDBCType jdbcType, Map<String, Object> dataTypeProperties) throws DataTypeConvertException;
+
+ /**
+ * Converts a connector data type to an EventMesh data type with additional data type properties.
+ *
+ * @param connectorDataType The connector data type to be converted.
+ * @param dataTypeProperties Additional data type properties.
+ * @return The converted EventMesh data type.
+ * @throws DataTypeConvertException If the conversion fails.
+ */
+ EventMeshDataType<?> toEventMeshType(T connectorDataType, Map<String, Object> dataTypeProperties) throws DataTypeConvertException;
+
+ /**
+ * Converts an EventMesh data type to a connector data type with additional data type properties.
+ *
+ * @param eventMeshDataType The EventMesh data type to be converted.
+ * @param dataTypeProperties Additional data type properties.
+ * @return The converted connector data type.
+ * @throws DataTypeConvertException If the conversion fails.
+ */
+ T toConnectorType(EventMeshDataType<?> eventMeshDataType, Map<String, Object> dataTypeProperties) throws DataTypeConvertException;
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DatabaseDialect.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DatabaseDialect.java
new file mode 100644
index 000000000..76dbf5fab
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/DatabaseDialect.java
@@ -0,0 +1,74 @@
+/*
+ * 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 org.apache.eventmesh.connector.jdbc.connection.JdbcConnection;
+import org.apache.eventmesh.connector.jdbc.connection.JdbcConnectionProvider;
+import org.apache.eventmesh.connector.jdbc.table.catalog.Catalog;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+/**
+ * Interface for a database dialect, which extends the ConnectionProvider and Catalog interfaces.
+ */
+public interface DatabaseDialect<JC extends JdbcConnection> extends JdbcConnectionProvider<JC>, Catalog {
+
+ /**
+ * Initializes the database dialect.
+ */
+ void init();
+
+ /**
+ * Starts the database dialect.
+ */
+ void start();
+
+ /**
+ * Retrieves the name of the database dialect.
+ *
+ * @return The name of the database dialect.
+ */
+ String getName();
+
+ /**
+ * Creates a prepared statement for the given SQL query using the provided database connection.
+ *
+ * @param connection The database connection.
+ * @param sql The SQL query.
+ * @return The prepared statement.
+ * @throws SQLException If an error occurs while creating the prepared statement.
+ */
+ PreparedStatement createPreparedStatement(Connection connection, String sql) throws SQLException;
+
+ /**
+ * Retrieves the JDBC driver meta-data associated with the database dialect.
+ *
+ * @return The JDBC driver meta-data.
+ */
+ JdbcDriverMetaData getJdbcDriverMetaData();
+
+ /**
+ * Retrieves the JDBC protocol associated with the database dialect.
+ *
+ * @return The JDBC protocol.
+ */
+ String jdbcProtocol();
+}
+
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Field.java
similarity index 55%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Field.java
index 1de315b97..24dcb5774 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Field.java
@@ -15,32 +15,11 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+public class Field {
+ private String type;
+ private String optional;
+ private String name;
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
similarity index 57%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
index 1de315b97..f70819212 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcConnectData.java
@@ -15,32 +15,27 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
+public final class JdbcConnectData {
+
+ private Payload payload = new Payload();
+
+ private Schema schema;
+
+ public Payload getPayload() {
+ return payload;
}
- public String getExtensionTypeName() {
- return extensionTypeName;
+ 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-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcContext.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcContext.java
index 1de315b97..506e4baa2 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcContext.java
@@ -15,32 +15,15 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Interface representing a JDBC context.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+public interface JdbcContext<Part extends Partition, OffSetCtx extends OffsetContext> {
- OFFSETMGMT("offsetMgmt"),
- ;
+ Part getPartition();
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ OffSetCtx getOffsetContext();
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcDriverMetaData.java
similarity index 54%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcDriverMetaData.java
index 1de315b97..fdc1e0fcd 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/JdbcDriverMetaData.java
@@ -15,32 +15,33 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.ToString;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Represents metadata information about a JDBC driver
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+@Data
+@AllArgsConstructor
+@ToString
+public class JdbcDriverMetaData {
+
+ // The major version number of the JDBC driver
+ private final int jdbcMajorVersion;
+
+ // The minor version number of the JDBC driver
+ private final int jdbcMinorVersion;
+
+ // The name of the JDBC driver
+ private final String jdbcDriverName;
+
+ // The name of the database product
+ private final String databaseProductName;
+
+ // The version of the database product
+ private final String databaseProductVersion;
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/OffsetContext.java
similarity index 53%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/OffsetContext.java
index 1de315b97..98bd2213d 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/OffsetContext.java
@@ -15,32 +15,35 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc;
+
+import java.util.Map;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * The OffsetContext interface represents the offset context for event processing. It provides methods to retrieve the offset map and check if a
+ * snapshot is currently running.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+public interface OffsetContext {
+
+ /**
+ * Retrieves the offset map associated with the context.
+ *
+ * @return The offset map.
+ */
+ Map<String, ?> getOffset();
- OFFSETMGMT("offsetMgmt"),
- ;
+ /**
+ * Checks if a snapshot is currently running.
+ *
+ * @return True if a snapshot is running, false otherwise.
+ */
+ boolean isSnapshotRunning();
- private final String extensionTypeName;
+ boolean isSnapshotCompleted();
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ void markSnapshotRunning();
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ void markSnapshotCompleted();
}
+
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Partition.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Partition.java
index 1de315b97..e1394a4f7 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Partition.java
@@ -15,32 +15,21 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc;
+
+import java.util.Map;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * This interface represents a partition.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+public interface Partition {
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * <p>Returns a map representing the partition.The keys of the map represent the partition keys, and the values represent the corresponding
+ * partition values.<p>
+ *
+ * @return a map representing the partition
+ */
+ Map<String, String> getPartition();
}
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
new file mode 100644
index 000000000..2a9235bd9
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Payload.java
@@ -0,0 +1,72 @@
+/*
+ * 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 org.apache.eventmesh.connector.jdbc.source.SourceMateData;
+
+import java.util.HashMap;
+
+public final class Payload extends HashMap<String, Object> {
+
+ public Payload withSource(SourceMateData source) {
+ super.put("source", source);
+ return this;
+ }
+
+ public Payload withDdl(String ddl) {
+ super.put("ddl", ddl);
+ return this;
+ }
+
+ public Payload withCatalogChanges(CatalogChanges catalogChanges) {
+ super.put("catalogChanges", catalogChanges);
+ return this;
+ }
+
+ public SourceMateData ofSourceMateData() {
+ return (SourceMateData) super.get("source");
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static class Builder {
+
+ private final Payload payload;
+
+ private Builder() {
+ payload = new Payload();
+ }
+
+ public Builder put(String key, Object value) {
+ payload.put(key, value);
+ return this;
+ }
+
+ public Builder withSource(SourceMateData source) {
+ payload.put("source", source);
+ return this;
+ }
+
+ public Payload build() {
+ return payload;
+ }
+ }
+
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
index 1de315b97..7916a911c 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/Schema.java
@@ -15,32 +15,16 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+import java.util.List;
- OFFSETMGMT("offsetMgmt"),
- ;
+public class Schema {
- private final String extensionTypeName;
+ private boolean optional;
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ private Field field;
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ private List<Field> fields;
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/config/JdbcConfig.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/config/JdbcConfig.java
new file mode 100644
index 000000000..9c7558426
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/config/JdbcConfig.java
@@ -0,0 +1,63 @@
+/*
+ * 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.config;
+
+import java.util.Properties;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Represents the configuration for a JDBC connection.
+ */
+@AllArgsConstructor
+@NoArgsConstructor
+@Data
+public class JdbcConfig {
+
+ private String databaseName;
+
+ // The hostname of the database server.
+ private String hostname;
+
+ // The port number of the database server.
+ private int port;
+
+ // The username for the database connection.
+ private String user;
+
+ // The password for the database connection.
+ private String password;
+
+ private String initialStatements;
+
+ private int connectTimeout;
+
+ /**
+ * Converts the JdbcConfig object to a Properties object containing the user and password.
+ *
+ * @return The Properties object representing the JdbcConfig.
+ */
+ public Properties asProperties() {
+ Properties props = new Properties();
+ props.put("user", this.user);
+ props.put("password", this.password);
+ return props;
+ }
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/config/JdbcServerConfig.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/config/JdbcServerConfig.java
index 1de315b97..451ecf71e 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/config/JdbcServerConfig.java
@@ -15,32 +15,24 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.config;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+import org.apache.eventmesh.openconnect.api.config.Config;
- OFFSETMGMT("offsetMgmt"),
- ;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
- private final String extensionTypeName;
+/**
+ * configuration for the JDBC server.
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class JdbcServerConfig extends Config {
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ // Indicates whether the source is enabled or not
+ private boolean sourceEnable;
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ // Indicates whether the sink is enabled or not
+ private boolean sinkEnable;
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnection.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnection.java
new file mode 100644
index 000000000..373927b09
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnection.java
@@ -0,0 +1,596 @@
+/*
+ * 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.connection;
+
+import org.apache.eventmesh.connector.jdbc.JdbcDriverMetaData;
+import org.apache.eventmesh.connector.jdbc.config.JdbcConfig;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+/**
+ * JdbcConnection class representing a JDBC connection.
+ * Implements the AutoCloseable interface.
+ */
+public class JdbcConnection implements AutoCloseable {
+
+ private static final int CONNECTION_VALID_CHECK_TIMEOUT_IN_SEC = 3;
+
+ private static final String STATEMENT_DELIMITER = ";";
+
+ private final JdbcConfig jdbcConfig;
+
+ private volatile Connection connection;
+
+ private final InitialOperation initialOperation;
+
+ private final ConnectionFactory connectionFactory;
+
+ private JdbcDriverMetaData jdbcDriverMetaData;
+
+ private boolean lazyConnection = true;
+
+ public JdbcConnection(JdbcConfig jdbcConfig, InitialOperation initialOperation, ConnectionFactory connectionFactory) {
+ this(jdbcConfig, initialOperation, connectionFactory, true);
+ }
+
+ public JdbcConnection(JdbcConfig jdbcConfig, InitialOperation initialOperation, ConnectionFactory connectionFactory, boolean lazyConnection) {
+ this.jdbcConfig = jdbcConfig;
+ this.initialOperation = initialOperation;
+ this.connectionFactory = connectionFactory;
+ this.lazyConnection = lazyConnection;
+ if (!this.lazyConnection) {
+ try {
+ connection();
+ } catch (SQLException e) {
+ log.warn("Get Connection error", e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /**
+ * Closes the JDBC connection.
+ *
+ * @throws Exception if an error occurs while closing the connection.
+ */
+ @Override
+ public void close() throws Exception {
+ if (connection != null) {
+ connection.close();
+ }
+ }
+
+ /**
+ * Retrieves the JDBC configuration.
+ *
+ * @return The JDBC configuration.
+ */
+ public JdbcConfig getJdbcConfig() {
+ return jdbcConfig;
+ }
+
+ /**
+ * Sets the auto-commit mode for the connection.
+ *
+ * @param autoCommit The auto-commit mode.
+ * @return The JdbcConnection instance.
+ * @throws SQLException if a database access error occurs.
+ */
+ public JdbcConnection setAutoCommit(boolean autoCommit) throws SQLException {
+ connection().setAutoCommit(autoCommit);
+ return this;
+ }
+
+ /**
+ * Retrieves the JDBC connection. Creates a new connection if not already connected.
+ *
+ * @return The JDBC connection.
+ * @throws SQLException if a database access error occurs.
+ */
+ public synchronized Connection connection() throws SQLException {
+ return connection(true);
+ }
+
+ /**
+ * Retrieves the JDBC connection. Creates a new connection if not already connected.
+ *
+ * @param executeOnConnect Flag indicating whether to execute initial statements on connect.
+ * @return The JDBC connection.
+ * @throws SQLException if a database access error occurs.
+ */
+ public synchronized Connection connection(boolean executeOnConnect) throws SQLException {
+ if (!isConnected()) {
+ connection = connectionFactory.connect(jdbcConfig);
+ if (!isConnected()) {
+ throw new SQLException("Unable to obtain a JDBC connection");
+ }
+
+ if (initialOperation != null) {
+ execute(initialOperation);
+ }
+ final String statements = jdbcConfig.getInitialStatements();
+ if (StringUtils.isNotBlank(statements) && executeOnConnect) {
+ String[] split = statements.split(STATEMENT_DELIMITER);
+ execute(split);
+ }
+ jdbcDriverMetaData = createJdbcDriverInfo();
+ }
+ return connection;
+ }
+
+ /**
+ * Creates the JDBC driver metadata by retrieving information from the provided connection's database metadata.
+ *
+ * @return The JDBC driver metadata.
+ * @throws SQLException if a database access error occurs.
+ */
+ private JdbcDriverMetaData createJdbcDriverInfo() throws SQLException {
+ DatabaseMetaData metadata = connection.getMetaData();
+
+ // Retrieve the JDBC driver information from the database metadata
+ int majorVersion = metadata.getJDBCMajorVersion();
+ int minorVersion = metadata.getJDBCMinorVersion();
+ String driverName = metadata.getDriverName();
+ String productName = metadata.getDatabaseProductName();
+ String productVersion = metadata.getDatabaseProductVersion();
+
+ // Create and return the JdbcDriverMetaData instance
+ return new JdbcDriverMetaData(majorVersion, minorVersion, driverName, productName, productVersion);
+ }
+
+
+ public JdbcDriverMetaData getJdbcDriverMetaData() {
+ return jdbcDriverMetaData;
+ }
+
+ /**
+ * Executes SQL statements on the JDBC connection.
+ *
+ * @param sqlStatements The SQL statements to execute.
+ * @return The JdbcConnection instance.
+ * @throws SQLException if a database access error occurs.
+ */
+ public JdbcConnection execute(String... sqlStatements) throws SQLException {
+ return execute(statement -> {
+ for (String sqlStatement : sqlStatements) {
+ if (sqlStatement != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Executing '{}'", sqlStatement);
+ }
+ statement.execute(sqlStatement);
+ }
+ }
+ });
+ }
+
+ /**
+ * Executes a custom initial operation on the JDBC connection.
+ *
+ * @param operation The initial operation to execute.
+ * @return The JdbcConnection instance.
+ * @throws SQLException if a database access error occurs.
+ */
+ public JdbcConnection execute(InitialOperation operation) throws SQLException {
+ Connection conn = connection();
+ try (Statement statement = conn.createStatement()) {
+ operation.apply(statement);
+ commit();
+ }
+ return this;
+ }
+
+ /**
+ * Execute the given SQL statements without committing the changes.
+ *
+ * @param sqlStatements The SQL statements to execute
+ * @return This JdbcConnection instance
+ * @throws SQLException If an SQL error occurs
+ */
+ public JdbcConnection executeWithoutCommitting(String... sqlStatements) throws SQLException {
+ Connection conn = connection();
+ if (conn.getAutoCommit()) {
+ throw new SQLException("Cannot execute without committing because auto-commit is enabled");
+ }
+
+ try (Statement statement = conn.createStatement()) {
+ for (String sqlStatement : sqlStatements) {
+ if (log.isDebugEnabled()) {
+ log.debug("Executing sql statement: {}", sqlStatement);
+ }
+ statement.execute(sqlStatement);
+ }
+ }
+
+ return this;
+ }
+
+ /**
+ * Checks if the JDBC connection is connected.
+ *
+ * @return true if the connection is connected, false otherwise.
+ * @throws SQLException if a database access error occurs.
+ */
+ public synchronized boolean isConnected() throws SQLException {
+ if (connection == null) {
+ return false;
+ }
+ return !connection.isClosed();
+ }
+
+ /**
+ * Checks if the JDBC connection is valid.
+ *
+ * @return true if the connection is valid, false otherwise.
+ * @throws SQLException if a database access error occurs.
+ */
+ public synchronized boolean isValid() throws SQLException {
+ return isConnected() && connection.isValid(CONNECTION_VALID_CHECK_TIMEOUT_IN_SEC);
+ }
+
+ /**
+ * Commits the changes on the JDBC connection.
+ *
+ * @return The JdbcConnection instance.
+ * @throws SQLException if a database access error occurs.
+ */
+ public JdbcConnection commit() throws SQLException {
+ Connection conn = connection();
+ if (!conn.getAutoCommit()) {
+ conn.commit();
+ }
+ return this;
+ }
+
+ /**
+ * Executes a query on the JDBC connection and consumes the result set using the provided consumer.
+ *
+ * @param sql The SQL query to execute.
+ * @param resultConsumer The consumer to process the result set.
+ * @return The JdbcConnection instance.
+ * @throws SQLException if a database access error occurs.
+ */
+ public JdbcConnection query(String sql, JdbcResultSetConsumer resultConsumer) throws SQLException {
+ // Check if the connection is connected and valid
+ if (isConnected() && isValid()) {
+ connection();
+ }
+ return query(sql, Connection::createStatement, resultConsumer);
+ }
+
+ /**
+ * Executes a query on the JDBC connection and consumes the result set using the provided consumer.
+ *
+ * @param sql The SQL query to execute.
+ * @param statementFactory The factory to create the statement.
+ * @param resultConsumer The consumer to process the result set.
+ * @return The JdbcConnection instance.
+ * @throws SQLException if a database access error occurs.
+ */
+ public JdbcConnection query(String sql, StatementFactory statementFactory, JdbcResultSetConsumer resultConsumer) throws SQLException {
+ Connection conn = connection();
+ try (Statement statement = statementFactory.createStatement(conn)) {
+ if (log.isDebugEnabled()) {
+ log.debug("Query sql '{}'", sql);
+ }
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ if (resultConsumer != null) {
+ resultConsumer.accept(resultSet);
+ }
+ }
+ }
+ return this;
+ }
+
+ /**
+ * Executes a query on the JDBC connection and maps the result set using the provided ResultSetMapper.
+ *
+ * @param sql The SQL query to execute.
+ * @param resultSetMapper The mapper to map the result set to an object.
+ * @return The mapped object.
+ * @throws SQLException if a database access error occurs.
+ */
+ public <T> T query(String sql, ResultSetMapper<T> resultSetMapper) throws SQLException {
+ // Check if the connection is connected and valid
+ if (isConnected() && isValid()) {
+ connection();
+ }
+ return query(sql, Connection::createStatement, resultSetMapper);
+ }
+
+ /**
+ * Executes a query on the JDBC connection and maps the result set using the provided ResultSetMapper.
+ *
+ * @param sql The SQL query to execute.
+ * @param statementFactory The factory to create the statement.
+ * @param resultSetMapper The mapper to map the result set to an object.
+ * @return The mapped object.
+ * @throws SQLException if a database access error occurs.
+ */
+ public <T> T query(String sql, StatementFactory statementFactory, ResultSetMapper<T> resultSetMapper) throws SQLException {
+ Connection conn = connection();
+ try (Statement statement = statementFactory.createStatement(conn)) {
+ if (log.isDebugEnabled()) {
+ log.debug("Query sql '{}'", sql);
+ }
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ if (resultSetMapper != null) {
+ return resultSetMapper.map(resultSet);
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Executes a prepared query on the JDBC connection and consumes the result set using the provided consumer.
+ *
+ * @param sql The SQL query to execute.
+ * @param resultConsumer The consumer to process the result set.
+ * @param preparedParameters The prepared parameters for the query.
+ * @return The JdbcConnection instance.
+ * @throws SQLException if a database access error occurs.
+ */
+ public JdbcConnection preparedQuery(String sql, JdbcResultSetConsumer resultConsumer, PreparedParameter... preparedParameters)
+ throws SQLException {
+ // Check if the connection is connected and valid
+ if (isConnected() && isValid()) {
+ connection();
+ }
+ return preparedQuery(sql, (conn, statement) -> conn.prepareStatement(sql), resultConsumer, preparedParameters);
+ }
+
+ /**
+ * Executes a prepared query on the JDBC connection and consumes the result set using the provided consumer.
+ *
+ * @param sql The SQL query to execute.
+ * @param preparedStatementFactory The factory to create the prepared statement.
+ * @param resultConsumer The consumer to process the result set.
+ * @param preparedParameters The prepared parameters for the query.
+ * @return The JdbcConnection instance.
+ * @throws SQLException if a database access error occurs.
+ */
+ public JdbcConnection preparedQuery(String sql, PreparedStatementFactory preparedStatementFactory, JdbcResultSetConsumer resultConsumer,
+ PreparedParameter... preparedParameters) throws SQLException {
+
+ Connection conn = connection();
+ try (PreparedStatement preparedStatement = preparedStatementFactory.createPreparedStatement(conn, sql)) {
+ if (log.isDebugEnabled()) {
+ log.debug("Query sql '{}'", sql);
+ }
+ if (preparedParameters != null) {
+ for (int index = 0; index < preparedParameters.length; ++index) {
+ final PreparedParameter preparedParameter = preparedParameters[index];
+ if (preparedParameter.getJdbcType() == null) {
+ preparedStatement.setObject(index + 1, preparedParameter.getValue());
+ } else {
+ preparedStatement.setObject(index + 1, preparedParameter.getValue(), preparedParameter.getJdbcType().getVendorTypeNumber());
+ }
+ }
+ }
+ try (ResultSet resultSet = preparedStatement.executeQuery()) {
+ if (resultConsumer != null) {
+ resultConsumer.accept(resultSet);
+ }
+ }
+ }
+ return this;
+ }
+
+
+ /**
+ * Executes a prepared query on the JDBC connection and maps the result set using the provided ResultSetMapper.
+ *
+ * @param sql The SQL query to execute.
+ * @param resultSetMapper The mapper to map the result set to an object.
+ * @param preparedParameters The prepared parameters for the query.
+ * @return The mapped object.
+ * @throws SQLException if a database access error occurs.
+ */
+ public <T> T preparedQuery(String sql, ResultSetMapper<T> resultSetMapper, PreparedParameter... preparedParameters)
+ throws SQLException {
+ // Check if the connection is connected and valid
+ if (isConnected() && isValid()) {
+ connection();
+ }
+ return preparedQuery(sql, (conn, statement) -> conn.prepareStatement(sql), resultSetMapper, preparedParameters);
+ }
+
+ /**
+ * Executes a prepared query on the JDBC connection and maps the result set using the provided ResultSetMapper.
+ *
+ * @param sql The SQL query to execute.
+ * @param preparedStatementFactory The factory to create the prepared statement.
+ * @param resultSetMapper The mapper to map the result set to an object.
+ * @param preparedParameters The prepared parameters for the query.
+ * @return The mapped object.
+ * @throws SQLException if a database access error occurs.
+ */
+ public <T> T preparedQuery(String sql, PreparedStatementFactory preparedStatementFactory, ResultSetMapper<T> resultSetMapper,
+ PreparedParameter... preparedParameters) throws SQLException {
+
+ Connection conn = connection();
+ try (PreparedStatement preparedStatement = preparedStatementFactory.createPreparedStatement(conn, sql)) {
+ if (log.isDebugEnabled()) {
+ log.debug("Query sql '{}'", sql);
+ }
+ if (preparedParameters != null) {
+ for (int index = 0; index < preparedParameters.length; ++index) {
+ final PreparedParameter preparedParameter = preparedParameters[index];
+ if (preparedParameter.getJdbcType() == null) {
+ preparedStatement.setObject(index + 1, preparedParameter.getValue());
+ } else {
+ preparedStatement.setObject(index + 1, preparedParameter.getValue(), preparedParameter.getJdbcType().getVendorTypeNumber());
+ }
+ }
+ }
+ try (ResultSet resultSet = preparedStatement.executeQuery()) {
+ if (resultSetMapper != null) {
+ return resultSetMapper.map(resultSet);
+ }
+ }
+ }
+ return null;
+ }
+
+ public Statement createStatement(int fetchSize, int defaultFetchSize) throws SQLException {
+ final Statement statement = connection().createStatement();
+ statement.setFetchSize(fetchSize <= 0 ? defaultFetchSize : fetchSize);
+ return statement;
+ }
+
+
+ /**
+ * Functional interface for the initial operation on the JDBC connection.
+ */
+ @FunctionalInterface
+ public interface InitialOperation {
+
+ /**
+ * Applies the operation on the JDBC statement.
+ *
+ * @param statement The JDBC statement.
+ * @throws SQLException if a database access error occurs.
+ */
+ void apply(Statement statement) throws SQLException;
+ }
+
+ /**
+ * Functional interface for creating JDBC statements.
+ */
+ @FunctionalInterface
+ public interface StatementFactory {
+
+ /**
+ * Creates a JDBC statement.
+ *
+ * @param connection The JDBC connection.
+ * @return The JDBC statement.
+ * @throws SQLException if a database access error occurs.
+ */
+ Statement createStatement(Connection connection) throws SQLException;
+ }
+
+ /**
+ * Functional interface for creating JDBC prepared statements.
+ */
+ @FunctionalInterface
+ public interface PreparedStatementFactory {
+
+ /**
+ * Creates a JDBC prepared statement with the provided connection and SQL query.
+ *
+ * @param connection The JDBC connection.
+ * @param sql The SQL query.
+ * @return The JDBC prepared statement.
+ * @throws SQLException if a database access error occurs.
+ */
+ PreparedStatement createPreparedStatement(Connection connection, String sql) throws SQLException;
+ }
+
+ /**
+ * Functional interface for creating JDBC connections.
+ */
+ @FunctionalInterface
+ @ThreadSafe
+ public interface ConnectionFactory {
+
+ /**
+ * Creates a JDBC connection.
+ *
+ * @param config The JDBC configuration.
+ * @return The JDBC connection.
+ * @throws SQLException if a database access error occurs.
+ */
+ Connection connect(JdbcConfig config) throws SQLException;
+ }
+
+ /**
+ * Functional interface for mapping a ResultSet to an object of type T.
+ *
+ * @param <T> The type of object to be mapped.
+ */
+ @FunctionalInterface
+ public interface ResultSetMapper<T> {
+
+ /**
+ * Maps a ResultSet to an object of type T.
+ *
+ * @param rs The ResultSet to be mapped.
+ * @return The mapped object.
+ * @throws SQLException if a database access error occurs.
+ */
+ T map(ResultSet rs) throws SQLException;
+ }
+
+ /**
+ * Functional interface for consuming a ResultSet.
+ */
+ @FunctionalInterface
+ public interface JdbcResultSetConsumer {
+
+ /**
+ * Accepts a ResultSet and performs an operation on it.
+ *
+ * @param resultSet The ResultSet to be consumed.
+ * @throws SQLException if a database access error occurs.
+ */
+ void accept(ResultSet resultSet) throws SQLException;
+
+ }
+
+ /**
+ * Creates a ConnectionFactory that uses a pattern-based URL with placeholder values.
+ *
+ * @param urlWithPlaceholder The URL pattern with placeholders.
+ * @param replaces The replacement values for the placeholders.
+ * @return The ConnectionFactory instance.
+ */
+ @SuppressWarnings("unchecked")
+ public static ConnectionFactory createPatternConnectionFactory(String urlWithPlaceholder, String... replaces) {
+ return config -> {
+ String url;
+ if (replaces != null && replaces.length > 0) {
+ url = String.format(urlWithPlaceholder, (Object[]) replaces);
+ } else {
+ url = urlWithPlaceholder;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("URL: {}", url);
+ }
+ Connection connection = DriverManager.getConnection(url, config.asProperties());
+ if (log.isDebugEnabled()) {
+ log.debug("User [{}] Connected to {}", config.getUser(), url);
+ }
+ return connection;
+ };
+ }
+
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnectionProvider.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnectionProvider.java
new file mode 100644
index 000000000..602208eb6
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/JdbcConnectionProvider.java
@@ -0,0 +1,50 @@
+/*
+ * 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.connection;
+
+import org.apache.eventmesh.connector.jdbc.exception.JdbcConnectionException;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/**
+ * Obtaining a database connection and checking its validity.
+ */
+public interface JdbcConnectionProvider<JC extends JdbcConnection> extends AutoCloseable {
+
+ /**
+ * Obtains a database connection.
+ *
+ * @return A database connection.
+ */
+ JC getConnection();
+
+ JC newConnection();
+
+ /**
+ * Checks if a database connection is valid.
+ *
+ * @param connection The database connection to check.
+ * @param timeout The timeout in seconds.
+ * @return True if the connection is valid, false otherwise.
+ * @throws JdbcConnectionException If there is an error checking the connection.
+ * @throws SQLException If there is an error with the SQL query.
+ */
+ boolean isValid(Connection connection, int timeout) throws JdbcConnectionException, SQLException;
+
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/PreparedParameter.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/PreparedParameter.java
index 1de315b97..4461e1041 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/connection/PreparedParameter.java
@@ -15,32 +15,20 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.connection;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+import java.sql.JDBCType;
- OFFSETMGMT("offsetMgmt"),
- ;
+import lombok.Data;
- private final String extensionTypeName;
+/**
+ * Represents a parameter for a prepared statement.
+ */
+@Data
+public class PreparedParameter {
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ private Object value;
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ private JDBCType jdbcType;
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/AbstractDdlParser.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/AbstractDdlParser.java
new file mode 100644
index 000000000..fdac829dd
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/AbstractDdlParser.java
@@ -0,0 +1,60 @@
+/*
+ * 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.ddl;
+
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
+
+import lombok.Getter;
+
+public abstract class AbstractDdlParser implements DdlParser {
+
+ private final boolean skipViews;
+
+ private final boolean skipComments;
+
+ @Getter
+ private final TableId tableId;
+
+ public AbstractDdlParser(boolean skipViews, boolean skipComments) {
+ this.skipViews = skipViews;
+ this.skipComments = skipComments;
+ this.tableId = new TableId();
+ }
+
+ @Override
+ public void setCurrentDatabase(String databaseName) {
+ this.tableId.setCatalogName(databaseName);
+ }
+
+ @Override
+ public void setCurrentSchema(String schema) {
+ this.tableId.setSchemaName(schema);
+ }
+
+ public String getCurrentDatabase() {
+ return this.tableId.getCatalogName();
+ }
+
+ public boolean isSkipViews() {
+ return skipViews;
+ }
+
+ public boolean isSkipComments() {
+ return skipComments;
+ }
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/DdlParser.java
similarity index 57%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/DdlParser.java
index 1de315b97..7fcb9dfd7 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/DdlParser.java
@@ -15,32 +15,30 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.ddl;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Interface for parsing DDL SQL statements.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+public interface DdlParser {
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
+ /**
+ * Parses the given DDL SQL statement.
+ *
+ * @param ddlSql the DDL SQL statement to be parsed
+ */
+ default void parse(String ddlSql) {
+ parse(ddlSql, null);
}
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ void parse(String ddlSql, DdlParserCallback callback);
+
+ /**
+ * Sets the current database.
+ *
+ * @param databaseName the name of the database to be set as current
+ */
+ void setCurrentDatabase(String databaseName);
+ void setCurrentSchema(String schema);
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/DdlParserCallback.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/DdlParserCallback.java
index 1de315b97..b69fdcab6 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/ddl/DdlParserCallback.java
@@ -15,32 +15,22 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.ddl;
+
+import org.apache.eventmesh.connector.jdbc.event.Event;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Functional interface for a DDL parser listener.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
+@FunctionalInterface
+public interface DdlParserCallback {
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Handles the DDL event.
+ *
+ * @param event The DDL event to handle.
+ */
+ void handle(Event event);
}
+
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DataChangeEvent.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DataChangeEvent.java
index 1de315b97..c1655c27f 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DataChangeEvent.java
@@ -15,32 +15,20 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.event;
+
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Represents a data change event, extending the common event interface.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
+public interface DataChangeEvent extends Event {
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Gets the type of data change event.
+ *
+ * @return The data change event type.
+ */
+ DataChangeEventType getDataChangeEventType();
}
+
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DataChangeEventType.java
similarity index 57%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DataChangeEventType.java
index 1de315b97..be79cb3eb 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/DataChangeEventType.java
@@ -15,32 +15,19 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.event;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
+public enum DataChangeEventType {
- private final String extensionTypeName;
+ INSERT("I"), UPDATE("U"), DELETE("D");
+ private final String code;
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
+ DataChangeEventType(String code) {
+ this.code = code;
}
- public String getExtensionTypeName() {
- return extensionTypeName;
+ public String ofCode() {
+ return this.code;
}
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/Event.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/Event.java
index 1de315b97..7b254f960 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/Event.java
@@ -15,32 +15,22 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.event;
+
+import org.apache.eventmesh.connector.jdbc.JdbcConnectData;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Top Event interface
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+public interface Event {
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Gets the table ID of the event.
+ *
+ * @return The table ID.
+ */
+ TableId getTableId();
+ JdbcConnectData getJdbcConnectData();
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/EventConsumer.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/EventConsumer.java
index 1de315b97..9ad6d453c 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/EventConsumer.java
@@ -15,32 +15,18 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.event;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Functional interface for consuming events.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+@FunctionalInterface
+public interface EventConsumer {
+ /**
+ * Accepts a snapshot event.
+ *
+ * @param event the snapshot event to be consumed
+ */
+ void accept(Event event);
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/EventHandler.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/EventHandler.java
index 1de315b97..cf50dfefc 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/EventHandler.java
@@ -15,32 +15,19 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+
+package org.apache.eventmesh.connector.jdbc.event;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Represents a handler for snapshot events.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+@FunctionalInterface
+public interface EventHandler {
+ /**
+ * Handles a snapshot event.
+ *
+ * @param event The SnapshotEvent to handle.
+ */
+ void handle(Event event);
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/SchemaChangeEvent.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/SchemaChangeEvent.java
index 1de315b97..ff1960837 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/SchemaChangeEvent.java
@@ -15,32 +15,19 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.event;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Represents a schema change event, extending the common event interface.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+public interface SchemaChangeEvent extends Event {
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ /**
+ * Gets the type of schema change event.
+ *
+ * @return The schema change event type.
+ */
+ SchemaChangeEventType getSchemaChangeEventType();
+}
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
-}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/SchemaChangeEventType.java
similarity index 57%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/SchemaChangeEventType.java
index 1de315b97..fbf51ef2e 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/event/SchemaChangeEventType.java
@@ -15,32 +15,30 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.event;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+public enum SchemaChangeEventType {
- OFFSETMGMT("offsetMgmt"),
+ DATABASE_CREATE("D", "C"),
+ DATABASE_DROP("D", "D"),
+ DATABASE_ALERT("D", "A"),
+ TABLE_CREATE("T", "C"),
+ TABLE_DROP("T", "D"),
+ TABLE_ALERT("T", "A"),
;
+ private final String type;
+ private final String operationType;
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
+ SchemaChangeEventType(String type, String operationType) {
+ this.type = type;
+ this.operationType = operationType;
}
- public String getExtensionTypeName() {
- return extensionTypeName;
+ public String ofType() {
+ return this.type;
}
+ public String ofOperationType() {
+ return this.operationType;
+ }
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/CatalogException.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/CatalogException.java
new file mode 100644
index 000000000..17fa873fb
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/CatalogException.java
@@ -0,0 +1,80 @@
+/*
+ * 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.exception;
+
+public class CatalogException extends RuntimeException {
+
+ /**
+ * Constructs a new runtime exception with {@code null} as its detail message. The cause is not initialized, and may subsequently be initialized
+ * by a call to {@link #initCause}.
+ */
+ public CatalogException() {
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message. The cause is not initialized, and may subsequently be initialized by a
+ * call to {@link #initCause}.
+ *
+ * @param message the detail message. The detail message is saved for later retrieval by the {@link #getMessage()} method.
+ */
+ public CatalogException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message and cause. <p>Note that the detail message associated with {@code cause}
+ * is
+ * <i>not</i> automatically incorporated in this runtime exception's detail message.
+ *
+ * @param message the detail message (which is saved for later retrieval by the {@link #getMessage()} method).
+ * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <code>null</code> value is permitted, and
+ * indicates that the cause is nonexistent or unknown.)
+ * @since 1.4
+ */
+ public CatalogException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified cause and a detail message of <code>(cause==null ? null : cause.toString())</code> (which
+ * typically contains the class and detail message of
+ * <code>cause</code>). This constructor is useful for runtime exceptions
+ * that are little more than wrappers for other throwables.
+ *
+ * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <code>null</code> value is permitted, and
+ * indicates that the cause is nonexistent or unknown.)
+ * @since 1.4
+ */
+ public CatalogException(Throwable cause) {
+ super(cause);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message, cause, suppression enabled or disabled, and writable stack trace enabled
+ * or disabled.
+ *
+ * @param message the detail message.
+ * @param cause the cause. (A {@code null} value is permitted, and indicates that the cause is nonexistent or unknown.)
+ * @param enableSuppression whether or not suppression is enabled or disabled
+ * @param writableStackTrace whether or not the stack trace should be writable
+ * @since 1.7
+ */
+ public CatalogException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/DataTypeConvertException.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/DataTypeConvertException.java
new file mode 100644
index 000000000..57b7d4f57
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/DataTypeConvertException.java
@@ -0,0 +1,82 @@
+/*
+ * 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.exception;
+
+
+public class DataTypeConvertException extends RuntimeException {
+
+ /**
+ * Constructs a new runtime exception with {@code null} as its detail message. The cause is not initialized, and may subsequently be initialized
+ * by a call to {@link #initCause}.
+ */
+ public DataTypeConvertException() {
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message. The cause is not initialized, and may subsequently be initialized by a
+ * call to {@link #initCause}.
+ *
+ * @param message the detail message. The detail message is saved for later retrieval by the {@link #getMessage()} method.
+ */
+ public DataTypeConvertException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message and cause. <p>Note that the detail message associated with {@code cause}
+ * is
+ * <i>not</i> automatically incorporated in this runtime exception's detail message.
+ *
+ * @param message the detail message (which is saved for later retrieval by the {@link #getMessage()} method).
+ * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <code>null</code> value is permitted, and
+ * indicates that the cause is nonexistent or unknown.)
+ * @since 1.4
+ */
+ public DataTypeConvertException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified cause and a detail message of <code>(cause==null ? null : cause.toString())</code> (which
+ * typically contains the class and detail message of
+ * <code>cause</code>). This constructor is useful for runtime exceptions
+ * that are little more than wrappers for other throwables.
+ *
+ * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <code>null</code> value is permitted, and
+ * indicates that the cause is nonexistent or unknown.)
+ * @since 1.4
+ */
+ public DataTypeConvertException(Throwable cause) {
+ super(cause);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message, cause, suppression enabled or disabled, and writable stack trace enabled
+ * or disabled.
+ *
+ * @param message the detail message.
+ * @param cause the cause. (A {@code null} value is permitted, and indicates that the cause is nonexistent or unknown.)
+ * @param enableSuppression whether or not suppression is enabled or disabled
+ * @param writableStackTrace whether or not the stack trace should be writable
+ * @since 1.7
+ */
+ public DataTypeConvertException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/DatabaseNotExistException.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/DatabaseNotExistException.java
new file mode 100644
index 000000000..5c363d28f
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/DatabaseNotExistException.java
@@ -0,0 +1,80 @@
+/*
+ * 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.exception;
+
+public class DatabaseNotExistException extends RuntimeException {
+
+ /**
+ * Constructs a new runtime exception with {@code null} as its detail message. The cause is not initialized, and may subsequently be initialized
+ * by a call to {@link #initCause}.
+ */
+ public DatabaseNotExistException() {
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message. The cause is not initialized, and may subsequently be initialized by a
+ * call to {@link #initCause}.
+ *
+ * @param message the detail message. The detail message is saved for later retrieval by the {@link #getMessage()} method.
+ */
+ public DatabaseNotExistException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message and cause. <p>Note that the detail message associated with {@code cause}
+ * is
+ * <i>not</i> automatically incorporated in this runtime exception's detail message.
+ *
+ * @param message the detail message (which is saved for later retrieval by the {@link #getMessage()} method).
+ * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <code>null</code> value is permitted, and
+ * indicates that the cause is nonexistent or unknown.)
+ * @since 1.4
+ */
+ public DatabaseNotExistException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified cause and a detail message of <code>(cause==null ? null : cause.toString())</code> (which
+ * typically contains the class and detail message of
+ * <code>cause</code>). This constructor is useful for runtime exceptions
+ * that are little more than wrappers for other throwables.
+ *
+ * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <code>null</code> value is permitted, and
+ * indicates that the cause is nonexistent or unknown.)
+ * @since 1.4
+ */
+ public DatabaseNotExistException(Throwable cause) {
+ super(cause);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message, cause, suppression enabled or disabled, and writable stack trace enabled
+ * or disabled.
+ *
+ * @param message the detail message.
+ * @param cause the cause. (A {@code null} value is permitted, and indicates that the cause is nonexistent or unknown.)
+ * @param enableSuppression whether or not suppression is enabled or disabled
+ * @param writableStackTrace whether or not the stack trace should be writable
+ * @since 1.7
+ */
+ public DatabaseNotExistException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/JdbcConnectionException.java
similarity index 55%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/JdbcConnectionException.java
index 1de315b97..dde57fc34 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/JdbcConnectionException.java
@@ -15,32 +15,8 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.exception;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+public class JdbcConnectionException extends RuntimeException {
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/TableNotExistException.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/TableNotExistException.java
new file mode 100644
index 000000000..dc5a58dc0
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/exception/TableNotExistException.java
@@ -0,0 +1,79 @@
+/*
+ * 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.exception;
+
+public class TableNotExistException extends RuntimeException {
+
+ /**
+ * Constructs a new runtime exception with {@code null} as its detail message. The cause is not initialized, and may subsequently be initialized
+ * by a call to {@link #initCause}.
+ */
+ public TableNotExistException() {
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message. The cause is not initialized, and may subsequently be initialized by a
+ * call to {@link #initCause}.
+ *
+ * @param message the detail message. The detail message is saved for later retrieval by the {@link #getMessage()} method.
+ */
+ public TableNotExistException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message and cause. <p>Note that the detail message associated with {@code cause}
+ * is <i>not</i> automatically incorporated in this runtime exception's detail message.
+ *
+ * @param message the detail message (which is saved for later retrieval by the {@link #getMessage()} method).
+ * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <code>null</code> value is permitted, and
+ * indicates that the cause is nonexistent or unknown.)
+ * @since 1.4
+ */
+ public TableNotExistException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified cause and a detail message of <code>(cause==null ? null : cause.toString())</code> (which
+ * typically contains the class and detail message of
+ * <code>cause</code>). This constructor is useful for runtime exceptions
+ * that are little more than wrappers for other throwables.
+ *
+ * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method). (A <code>null</code> value is permitted, and
+ * indicates that the cause is nonexistent or unknown.)
+ * @since 1.4
+ */
+ public TableNotExistException(Throwable cause) {
+ super(cause);
+ }
+
+ /**
+ * Constructs a new runtime exception with the specified detail message, cause, suppression enabled or disabled, and writable stack trace enabled
+ * or disabled.
+ *
+ * @param message the detail message.
+ * @param cause the cause. (A {@code null} value is permitted, and indicates that the cause is nonexistent or unknown.)
+ * @param enableSuppression whether or not suppression is enabled or disabled
+ * @param writableStackTrace whether or not the stack trace should be writable
+ * @since 1.7
+ */
+ public TableNotExistException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/server/JdbcConnectorServer.java
similarity index 51%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/server/JdbcConnectorServer.java
index 1de315b97..546478ce9 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/server/JdbcConnectorServer.java
@@ -15,32 +15,29 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.server;
+
+import org.apache.eventmesh.connector.jdbc.config.JdbcServerConfig;
+import org.apache.eventmesh.connector.jdbc.source.JdbcSourceConnector;
+import org.apache.eventmesh.openconnect.Application;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * JDBC connector server
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
+public class JdbcConnectorServer {
- private final String extensionTypeName;
+ public static void main(String[] args) throws Exception {
+ JdbcServerConfig serverConfig = ConfigUtil.parse(JdbcServerConfig.class, "server-config.yml");
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ if (serverConfig.isSourceEnable()) {
+ Application jdbcSourceApp = new Application();
+ jdbcSourceApp.run(JdbcSourceConnector.class);
+ }
- public String getExtensionTypeName() {
- return extensionTypeName;
+ if (serverConfig.isSinkEnable()) {
+ //TODO support
+ }
}
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/AbstractEventMeshJdbcEventTask.java
similarity index 53%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/AbstractEventMeshJdbcEventTask.java
index 1de315b97..3f7301fa7 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/AbstractEventMeshJdbcEventTask.java
@@ -15,32 +15,30 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
+import org.apache.eventmesh.common.ThreadWrapper;
+import org.apache.eventmesh.connector.jdbc.event.Event;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+public abstract class AbstractEventMeshJdbcEventTask extends ThreadWrapper implements EventMeshJdbcEventTask<Event> {
+
+ protected BlockingQueue<Event> eventBlockingQueue = new LinkedBlockingQueue<>(10000);
+
+ @Override
+ public void shutdown() {
+ super.shutdown();
}
- public String getExtensionTypeName() {
- return extensionTypeName;
+ @Override
+ public void close() throws Exception {
+ shutdown();
}
+ @Override
+ public void put(Event event) throws InterruptedException {
+ eventBlockingQueue.put(event);
+ }
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/AbstractJdbcTaskManager.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/AbstractJdbcTaskManager.java
new file mode 100644
index 000000000..a305bf53d
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/AbstractJdbcTaskManager.java
@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+
+import org.apache.eventmesh.connector.jdbc.source.config.JdbcSourceConfig;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+public abstract class AbstractJdbcTaskManager<Task extends EventMeshJdbcTask> implements JdbcTaskManager {
+
+ protected Map<TableId, Task> tableIdJdbcTaskMap = new ConcurrentHashMap<>();
+
+ protected Set<TableId> includeDatabaseTable;
+
+ protected JdbcSourceConfig jdbcSourceConfig;
+
+ protected List<Task> taskList = new ArrayList<>(128);
+
+ protected List<TaskManagerListener> listeners = new ArrayList<>(16);
+
+
+ @Override
+ public void start() {
+ taskList.forEach(EventMeshJdbcTask::start);
+ }
+
+ @Override
+ public void shutdown() {
+ taskList.forEach(EventMeshJdbcTask::shutdown);
+ }
+
+ @Override
+ public void close() throws Exception {
+ shutdown();
+ }
+
+ @Override
+ public void registerListener(TaskManagerListener listener) {
+ if (!Objects.isNull(listener)) {
+ listeners.add(listener);
+ }
+ }
+
+ public abstract Task select(TableId tableId);
+}
\ No newline at end of file
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/Engine.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/Engine.java
index 1de315b97..fa2cc73b4 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/Engine.java
@@ -15,32 +15,31 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
- OFFSETMGMT("offsetMgmt"),
- ;
+import java.util.Set;
- private final String extensionTypeName;
+/**
+ * Engine interface represents the core engine
+ */
+public interface Engine {
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ /**
+ * Initializes the engine.
+ */
+ void init();
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Starts the engine.
+ */
+ void start();
+ /**
+ * Retrieves the set of TableId objects representing the tables that the engine handles.
+ *
+ * @return The set of handled TableId objects.
+ */
+ Set<TableId> getHandledTables();
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventDispatcher.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventDispatcher.java
new file mode 100644
index 000000000..17e59335e
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventDispatcher.java
@@ -0,0 +1,50 @@
+/*
+ * 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;
+
+import org.apache.eventmesh.connector.jdbc.event.Event;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class EventDispatcher {
+
+ private SourceJdbcTaskManager sourceJdbcTaskManager;
+
+ public EventDispatcher(SourceJdbcTaskManager sourceJdbcTaskManager) {
+ this.sourceJdbcTaskManager = sourceJdbcTaskManager;
+ }
+
+ /**
+ * Dispatch CDC events.
+ *
+ * @param event The CDC event to be dispatched.
+ */
+ public void dispatch(Event event) {
+ TableId tableId = event.getTableId();
+ SourceEventMeshJdbcEventTask task = sourceJdbcTaskManager.select(tableId);
+ try {
+ // Put the CDC event into the selected JDBC task.
+ task.put(event);
+ } catch (InterruptedException e) {
+ log.warn("Dispatch CdcEvent error", e);
+ }
+ }
+
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventMeshJdbcEventTask.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventMeshJdbcEventTask.java
new file mode 100644
index 000000000..6b4c4d50d
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventMeshJdbcEventTask.java
@@ -0,0 +1,45 @@
+/*
+ * 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;
+
+import org.apache.eventmesh.connector.jdbc.event.Event;
+import org.apache.eventmesh.connector.jdbc.event.EventHandler;
+
+/**
+ * The EventMeshJdbcTask interface represents a task that interacts with the EventMesh through a JDBC connection. It extends the AutoCloseable
+ * interface, allowing the task to be managed efficiently.
+ */
+public interface EventMeshJdbcEventTask<E extends Event> extends EventMeshJdbcTask {
+
+ /**
+ * Puts an event into the task for processing.
+ *
+ * @param event The event to be processed.
+ * @throws InterruptedException If the operation is interrupted while waiting to put the event.
+ */
+ void put(E event) throws InterruptedException;
+
+ /**
+ * Registers a snapshot event handler to be executed when snapshot events occur.
+ *
+ * @param handler The SnapshotEventHandler to be registered.
+ */
+ void registerEventHandler(EventHandler handler);
+
+}
+
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventMeshJdbcTask.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventMeshJdbcTask.java
index 1de315b97..48830358b 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/EventMeshJdbcTask.java
@@ -15,32 +15,23 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * The EventMeshJdbcTask interface represents a task that interacts with the EventMesh through a JDBC connection. It extends the AutoCloseable
+ * interface, allowing the task to be managed efficiently.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+public interface EventMeshJdbcTask extends AutoCloseable {
- OFFSETMGMT("offsetMgmt"),
- ;
+ /**
+ * Starts the EventMesh JDBC task, initializing any necessary resources or connections.
+ */
+ void start();
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Shuts down the EventMesh JDBC task, releasing any acquired resources or connections.
+ */
+ void shutdown();
}
+
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcAllFactoryLoader.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcAllFactoryLoader.java
new file mode 100644
index 000000000..f3c09adb5
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcAllFactoryLoader.java
@@ -0,0 +1,72 @@
+/*
+ * 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;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.eventmesh.connector.jdbc.source.dialect.DatabaseDialectFactory;
+import org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngineFactory;
+import org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.SnapshotEngineFactory;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
+
+import java.util.Objects;
+
+import lombok.experimental.UtilityClass;
+
+/**
+ * Get a CdcEngineFactory for a given database name
+ */
+@UtilityClass
+public class JdbcAllFactoryLoader {
+
+ /**
+ * Returns a CdcEngineFactory for the given database name.
+ * <p>Throws NullPointerException if databaseName is null.</p>
+ * <p>Throws IllegalArgumentException if CdcEngineFactory is not supported for the given database name.</p>
+ *
+ * @param databaseName Name of the database for which CdcEngineFactory is required.
+ * @return CdcEngineFactory for the given database name.
+ */
+ public static CdcEngineFactory getCdcEngineFactory(String databaseName) {
+ checkNotNull(databaseName, "database name can not be null");
+ CdcEngineFactory engineFactory = EventMeshExtensionFactory.getExtension(CdcEngineFactory.class, databaseName);
+ return checkNotNull(engineFactory, "CdcEngineFactory: " + databaseName + " is not supported");
+ }
+
+ /**
+ * Returns a DatabaseDialectFactory based on the specified database name.
+ *
+ * @param databaseName the name of the database
+ * @return the DatabaseDialectFactory for the specified database name
+ * @throws NullPointerException if the database name is null
+ * @throws IllegalArgumentException if the specified database name is not supported
+ */
+ public static DatabaseDialectFactory getDatabaseDialectFactory(String databaseName) {
+ Objects.requireNonNull(databaseName, "database name can not be null");
+ DatabaseDialectFactory databaseDialectFactory = EventMeshExtensionFactory.getExtension(DatabaseDialectFactory.class, databaseName);
+ Objects.requireNonNull(databaseDialectFactory, "DatabaseDialectFactory: " + databaseName + " is not supported");
+ return databaseDialectFactory;
+ }
+
+ public static SnapshotEngineFactory getSnapshotEngineFactory(String databaseName) {
+ Objects.requireNonNull(databaseName, "database name can not be null");
+ SnapshotEngineFactory databaseDialectFactory = EventMeshExtensionFactory.getExtension(SnapshotEngineFactory.class, databaseName);
+ Objects.requireNonNull(databaseDialectFactory, "SnapshotEngineFactory: " + databaseName + " is not supported");
+ return databaseDialectFactory;
+ }
+}
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
new file mode 100644
index 000000000..d8147bf02
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnector.java
@@ -0,0 +1,207 @@
+/*
+ * 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;
+
+import org.apache.eventmesh.connector.jdbc.DatabaseDialect;
+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;
+import org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.SnapshotResult.SnapshotResultStatus;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+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 lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class JdbcSourceConnector extends SourceConnector {
+
+ private DatabaseDialect databaseDialect;
+
+ private CdcEngine cdcEngine;
+
+ private JdbcSourceConfig sourceConfig;
+
+ private EventDispatcher dispatcher;
+
+ private SourceJdbcTaskManager sourceJdbcTaskManager;
+
+ private SnapshotEngine<?> snapshotEngine;
+
+ private TaskManagerCoordinator taskManagerCoordinator;
+
+ public JdbcSourceConnector() {
+ this(null);
+ }
+
+ protected JdbcSourceConnector(SourceConfig sourceConfig) {
+ super(sourceConfig);
+ }
+
+ /**
+ * Returns the class type of the configuration for this Connector.
+ *
+ * @return Class type of the configuration
+ */
+ @Override
+ public Class<? extends Config> configClass() {
+ return JdbcSourceConfig.class;
+ }
+
+ /**
+ * Initializes the Connector with the provided configuration.
+ *
+ * @param config Configuration object
+ * @throws Exception if initialization fails
+ */
+ @Override
+ public void init(Config config) throws Exception {
+
+ if (!(config instanceof JdbcSourceConfig)) {
+ throw new IllegalArgumentException("Config not be JdbcSourceConfig");
+ }
+ this.sourceConfig = (JdbcSourceConfig) config;
+ doInit();
+ }
+
+ /**
+ * Initializes the Connector with the provided context.
+ *
+ * @param connectorContext connectorContext
+ * @throws Exception if initialization fails
+ */
+ @Override
+ public void init(ConnectorContext connectorContext) throws Exception {
+ SourceConnectorContext sourceConnectorContext = (SourceConnectorContext) connectorContext;
+ this.sourceConfig = (JdbcSourceConfig) sourceConnectorContext.getSourceConfig();
+ doInit();
+ }
+
+ private void doInit() {
+ String databaseType = this.sourceConfig.getSourceConnectorConfig().getDatabaseType();
+
+ // Get the database dialect factory and create the database dialect.
+ final DatabaseDialectFactory databaseDialectFactory = JdbcAllFactoryLoader.getDatabaseDialectFactory(databaseType);
+ this.databaseDialect = databaseDialectFactory.createDatabaseDialect(sourceConfig);
+ this.databaseDialect.init();
+
+ // Get the snapshot engine factory and create the snapshot engine
+ final SnapshotEngineFactory snapshotEngineFactory = JdbcAllFactoryLoader.getSnapshotEngineFactory(databaseType);
+ this.snapshotEngine = snapshotEngineFactory.createSnapshotEngine(this.sourceConfig, this.databaseDialect);
+ 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();
+
+ // Create the task manager and dispatcher.
+ this.sourceJdbcTaskManager = new SourceJdbcTaskManager(cdcEngine, this.sourceConfig);
+ this.sourceJdbcTaskManager.init();
+
+ this.dispatcher = new EventDispatcher(this.sourceJdbcTaskManager);
+
+ this.taskManagerCoordinator = new TaskManagerCoordinator();
+ this.taskManagerCoordinator.registerTaskManager(SourceJdbcTaskManager.class.getName(), sourceJdbcTaskManager);
+ this.taskManagerCoordinator.init();
+ }
+
+ private void eventConsumer(Event event) {
+ this.dispatcher.dispatch(event);
+ }
+
+ /**
+ * Starts the Connector.
+ *
+ * @throws Exception if the start operation fails
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ public void start() throws Exception {
+ this.databaseDialect.start();
+ this.taskManagerCoordinator.start();
+ this.snapshotEngine.start();
+ SnapshotResult<?> result = this.snapshotEngine.execute();
+ this.snapshotEngine.close();
+ //success and skip status can run cdc engine
+ if (result.getStatus() != SnapshotResultStatus.ABORTED) {
+ this.cdcEngine.setContext(result.getContext());
+ this.cdcEngine.start();
+ }
+ }
+
+ /**
+ * Commits the specified ConnectRecord object.
+ *
+ * @param record ConnectRecord object to commit
+ */
+ @Override
+ public void commit(ConnectRecord record) {
+
+ }
+
+ /**
+ * Returns the name of the Connector.
+ *
+ * @return String name of the Connector
+ */
+ @Override
+ public String name() {
+ return "JDBC Source Connector";
+ }
+
+ /**
+ * Stops the Connector.
+ *
+ * @throws Exception if stopping fails
+ */
+ @Override
+ public void stop() throws Exception {
+
+ }
+
+ @Override
+ public List<ConnectRecord> poll() {
+
+ List<ConnectRecord> connectRecords = this.taskManagerCoordinator.poll();
+
+ return connectRecords;
+ }
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcTaskManager.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcTaskManager.java
new file mode 100644
index 000000000..7bef16d9b
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/JdbcTaskManager.java
@@ -0,0 +1,49 @@
+/*
+ * 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;
+
+/**
+ * The JdbcTaskManager interface represents a manager for JDBC tasks. It extends the AutoCloseable interface, allowing the manager to be managed
+ * efficiently.
+ */
+public interface JdbcTaskManager extends AutoCloseable {
+
+ /**
+ * Initializes the JDBC task manager, setting up any required configurations or resources.
+ */
+ void init();
+
+ /**
+ * Starts the JDBC task manager, allowing it to begin managing tasks.
+ */
+ void start();
+
+ /**
+ * Shuts down the JDBC task manager, releasing any acquired resources or stopping task management.
+ */
+ void shutdown();
+
+ /**
+ * Registers a listener to receive events and notifications from the JDBC task manager.
+ *
+ * @param listener The listener to be registered.
+ */
+ void registerListener(TaskManagerListener listener);
+
+}
+
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
new file mode 100644
index 000000000..7c2b0d717
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceEventMeshJdbcEventTask.java
@@ -0,0 +1,77 @@
+/*
+ * 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;
+
+import org.apache.eventmesh.connector.jdbc.event.Event;
+import org.apache.eventmesh.connector.jdbc.event.EventHandler;
+
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+public class SourceEventMeshJdbcEventTask extends AbstractEventMeshJdbcEventTask {
+
+
+ private final String taskName;
+
+ private EventHandler eventHandler;
+
+ public SourceEventMeshJdbcEventTask(String taskName) {
+ this.taskName = taskName;
+
+ }
+
+ @Override
+ public String getThreadName() {
+ return taskName;
+ }
+
+ /**
+ * 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() {
+ while (isRunning) {
+ try {
+ Event snapshotEvent = eventBlockingQueue.poll(5, TimeUnit.SECONDS);
+ if (Objects.isNull(snapshotEvent)) {
+ continue;
+ }
+ eventHandler.handle(snapshotEvent);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+
+ /**
+ * Registers a snapshot event handler to be executed when snapshot events occur.
+ *
+ * @param handler The SnapshotEventHandler to be registered.
+ */
+ @Override
+ public void registerEventHandler(EventHandler handler) {
+ this.eventHandler = handler;
+ }
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceJdbcTaskManager.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceJdbcTaskManager.java
new file mode 100644
index 000000000..39243e3f9
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceJdbcTaskManager.java
@@ -0,0 +1,96 @@
+/*
+ * 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;
+
+import org.apache.eventmesh.connector.jdbc.JdbcConnectData;
+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.cdc.CdcEngine;
+import org.apache.eventmesh.connector.jdbc.source.dialect.cdc.RandomTaskSelectStrategy;
+import org.apache.eventmesh.connector.jdbc.source.dialect.cdc.TaskSelectStrategy;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+
+import org.apache.commons.collections4.CollectionUtils;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SourceJdbcTaskManager extends AbstractJdbcTaskManager<SourceEventMeshJdbcEventTask> {
+
+ private final Set<TableId> includeDatabaseTable;
+
+ private final JdbcSourceConfig jdbcSourceConfig;
+
+ private TaskSelectStrategy<SourceEventMeshJdbcEventTask> cdcTaskSelectStrategy;
+
+ public SourceJdbcTaskManager(CdcEngine cdcEngine, JdbcSourceConfig jdbcSourceConfig) {
+ this.jdbcSourceConfig = jdbcSourceConfig;
+ this.includeDatabaseTable =
+ CollectionUtils.isEmpty(cdcEngine.getHandledTables()) ? new HashSet<>() : new HashSet<>(cdcEngine.getHandledTables());
+ }
+
+
+ @SuppressWarnings("unchecked")
+ public void init() {
+ //init Jdbc Task
+ int maxTaskNum = this.jdbcSourceConfig.getSourceConnectorConfig().getMaxTask();
+ int taskNum = Math.min(maxTaskNum, this.includeDatabaseTable.size());
+ log.info("Source jdbc task num {}", taskNum);
+ for (int index = 0; index < taskNum; ++index) {
+ SourceEventMeshJdbcEventTask eventTask = new SourceEventMeshJdbcEventTask("source-jdbc-task-" + (index + 1));
+ eventTask.registerEventHandler(this::doHandleEvent);
+ taskList.add(eventTask);
+ }
+ cdcTaskSelectStrategy = new RandomTaskSelectStrategy(taskList);
+
+ }
+
+ private void doHandleEvent(Event event) {
+
+ JdbcConnectData jdbcConnectData = event.getJdbcConnectData();
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ ConnectRecord record = new ConnectRecord(partition, offset, System.currentTimeMillis(), jdbcConnectData);
+ if (null == record) {
+ return;
+ }
+ List<ConnectRecord> records = Arrays.asList(record);
+ for (TaskManagerListener listener : listeners) {
+ listener.listen(records);
+ }
+ }
+
+
+ @Override
+ public SourceEventMeshJdbcEventTask select(TableId tableId) {
+ return tableIdJdbcTaskMap.computeIfAbsent(tableId, key -> cdcTaskSelectStrategy.select(tableId));
+ }
+
+ public int getTaskCount() {
+ return tableIdJdbcTaskMap.size();
+ }
+
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceMateData.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceMateData.java
new file mode 100644
index 000000000..732b12524
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/SourceMateData.java
@@ -0,0 +1,64 @@
+/*
+ * 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;
+
+import lombok.Data;
+
+/**
+ * Represents metadata related to a data source.
+ */
+@Data
+public class SourceMateData {
+
+ /**
+ * The connector used for the connector source. e.g: mysql, oracle etc.
+ */
+ private String connector;
+
+ /**
+ * The name of the connector source.
+ */
+ private String name;
+
+ /**
+ * The timestamp when the metadata was captured.
+ */
+ private long timestamp;
+
+ /**
+ * Flag indicating whether this metadata belongs to a snapshot.
+ */
+ private boolean snapshot;
+
+ /**
+ * The catalog name associated with the connector source.
+ */
+ private String catalogName;
+
+ /**
+ * The schema name associated with the connector source.
+ */
+ private String schemaName;
+
+ /**
+ * The table name associated with the connector source.
+ */
+ private String tableName;
+
+}
+
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/TaskManagerCoordinator.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/TaskManagerCoordinator.java
new file mode 100644
index 000000000..b22dafa8e
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/TaskManagerCoordinator.java
@@ -0,0 +1,116 @@
+/*
+ * 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;
+
+
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+
+import org.apache.commons.collections4.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import lombok.extern.slf4j.Slf4j;
+
+
+/**
+ * The TaskManagerCoordinator is responsible for coordinating multiple JDBC task managers and managing the processing of ConnectRecords. It provides
+ * methods for registering task managers, initializing them, and starting their processing.
+ */
+@Slf4j
+public class TaskManagerCoordinator {
+
+ private static final int BATCH_MAX = 10;
+ private static final int DEFAULT_QUEUE_SIZE = 1 << 13;
+
+ private BlockingQueue<ConnectRecord> recordBlockingQueue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_SIZE);
+ private Map<String, JdbcTaskManager> taskManagerCache = new HashMap<>(8);
+
+ /**
+ * Constructs a new TaskManagerCoordinator.
+ */
+ public TaskManagerCoordinator() {
+ }
+
+ /**
+ * Registers a JDBC task manager with the given name.
+ *
+ * @param name The name of the task manager.
+ * @param taskManager The JDBC task manager to register.
+ */
+ public void registerTaskManager(String name, JdbcTaskManager taskManager) {
+ taskManagerCache.put(name, taskManager);
+ }
+
+ /**
+ * Initializes all registered JDBC task managers.
+ */
+ public void init() {
+ taskManagerCache.values().forEach(JdbcTaskManager::init);
+
+ // Register a listener on each task manager to process incoming records and add them to the blocking queue.
+ taskManagerCache.values().forEach(taskManager -> taskManager.registerListener(records -> {
+ if (CollectionUtils.isEmpty(records)) {
+ return;
+ }
+ records.forEach(record -> {
+ try {
+ recordBlockingQueue.put(record);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }));
+ }
+
+ /**
+ * Starts the processing of all registered JDBC task managers.
+ */
+ public void start() {
+ taskManagerCache.values().forEach(JdbcTaskManager::start);
+ }
+
+ /**
+ * Polls for a batch of ConnectRecords from the blocking queue.
+ *
+ * @return A list of ConnectRecords, up to the maximum batch size defined by BATCH_MAX.
+ */
+ public List<ConnectRecord> poll() {
+ List<ConnectRecord> records = new ArrayList<>(BATCH_MAX);
+ for (int index = 0; index < BATCH_MAX; ++index) {
+ try {
+ ConnectRecord record = recordBlockingQueue.poll(3, TimeUnit.SECONDS);
+ if (Objects.isNull(record)) {
+ break;
+ }
+ records.add(record);
+ } catch (InterruptedException e) {
+ break;
+ }
+ }
+ return records;
+ }
+}
+
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/TaskManagerListener.java
similarity index 55%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/TaskManagerListener.java
index 1de315b97..abfbfcc5d 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/TaskManagerListener.java
@@ -15,32 +15,24 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
- private final String extensionTypeName;
+import java.util.List;
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+/**
+ * The TaskManagerListener is a functional interface used to listen for events from the TaskManager. It defines a single method, "listen", that takes
+ * a list of ConnectRecord objects as its parameter.
+ */
+@FunctionalInterface
+public interface TaskManagerListener {
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Listens for events from the TaskManager and processes the given list of ConnectRecords.
+ *
+ * @param records The list of ConnectRecord objects to be processed.
+ */
+ void listen(List<ConnectRecord> records);
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/JdbcSourceConfig.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/JdbcSourceConfig.java
index 1de315b97..b330c331b 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/JdbcSourceConfig.java
@@ -15,32 +15,16 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source.config;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+import lombok.Data;
+import lombok.EqualsAndHashCode;
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class JdbcSourceConfig extends SourceConfig {
+ private SourceConnectorConfig sourceConnectorConfig;
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/MysqlConfig.java
similarity index 54%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/MysqlConfig.java
index 1de315b97..032921350 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/MysqlConfig.java
@@ -15,32 +15,40 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source.config;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+import lombok.Data;
+
+@Data
+public class MysqlConfig {
+
+ private int serverId;
+
+ private boolean keepAlive = true;
+
+ private long keepAliveInterval;
+
+ private SnapshotLockingMode snapshotLockingMode = SnapshotLockingMode.MINIMAL;
+
+ private boolean useGlobalLock = true;
+
+ public enum SnapshotLockingMode {
+
+ EXTENDED("extended"),
+
+ MINIMAL("minimal"),
+
+ NONE("none");
+
+ private final String value;
+
+ SnapshotLockingMode(String value) {
+ this.value = value;
+ }
- public String getExtensionTypeName() {
- return extensionTypeName;
+ public boolean usesLocking() {
+ return !value.equals(NONE.value);
+ }
}
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/SourceConnectorConfig.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/SourceConnectorConfig.java
new file mode 100644
index 000000000..99210da8e
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/config/SourceConnectorConfig.java
@@ -0,0 +1,78 @@
+/*
+ * 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.config;
+
+import org.apache.eventmesh.connector.jdbc.config.JdbcConfig;
+
+import java.util.List;
+
+import lombok.Data;
+
+/**
+ * Represents the configuration for a database connector.
+ */
+@Data
+public class SourceConnectorConfig {
+
+ private static final int DEFAULT_SNAPSHOT_FETCH_SIZE = 100;
+
+ /**
+ * Max task number,The maximum cannot exceed the number of tables scanned. If it exceeds, it will be set to the number of tables.
+ */
+ private int maxTask;
+
+ private int batchMaxRows;
+
+ private boolean skipSnapshot = false;
+
+ // A list of database names to include in the connector.
+ private List<String> databaseIncludeList;
+
+ // A list of database names to exclude from the connector.
+ private List<String> databaseExcludeList;
+
+ // A list of table names to include in the connector.
+ private List<String> tableIncludeList;
+
+ // A list of table names to exclude from the connector.
+ private List<String> tableExcludeList;
+
+ // database type e.g. mysql
+ private String databaseType;
+
+ private int snapshotMaxThreads;
+
+ //The snapshot mode also require handling the database schema (database and table schema)
+ private boolean snapshotSchema = true;
+
+ //The snapshot mode require handling table data
+ private boolean snapshotData = true;
+
+ private int snapshotFetchSize = DEFAULT_SNAPSHOT_FETCH_SIZE;
+
+ private JdbcConfig jdbcConfig;
+
+ private boolean skipViews = false;
+
+ private boolean skipComments = false;
+
+ // The configuration for the MySQL database.
+ private MysqlConfig mysqlConfig;
+
+ private String name = "mysql-connector";
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/DatabaseDialectFactory.java
similarity index 50%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/DatabaseDialectFactory.java
index 1de315b97..8f73f9f53 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/DatabaseDialectFactory.java
@@ -15,32 +15,25 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source.dialect;
+
+import org.apache.eventmesh.connector.jdbc.DatabaseDialect;
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
+import org.apache.eventmesh.spi.EventMeshExtensionType;
+import org.apache.eventmesh.spi.EventMeshSPI;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Interface for creating a database dialect based on the provided source configuration.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.JDBC_DATABASE_DIALECT)
+public interface DatabaseDialectFactory {
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Creates a database dialect based on the provided source configuration.
+ *
+ * @param config the source configuration to create a database dialect for
+ * @return the created database dialect
+ */
+ DatabaseDialect createDatabaseDialect(SourceConfig config);
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/CdcEngine.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/CdcEngine.java
new file mode 100644
index 000000000..566ffccf4
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/CdcEngine.java
@@ -0,0 +1,50 @@
+/*
+ * 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.connector.jdbc.JdbcContext;
+import org.apache.eventmesh.connector.jdbc.event.EventConsumer;
+import org.apache.eventmesh.connector.jdbc.source.Engine;
+
+/**
+ * CdcEngine is a service that captures data change events.
+ */
+public interface CdcEngine<Context extends JdbcContext> extends Engine, AutoCloseable {
+
+ /**
+ * Stops the CDC Engine.
+ */
+ @Override
+ void close() throws Exception;
+
+ /**
+ * Returns the name of the CDC Engine.
+ *
+ * @return String representing the name of the CDC Engine.
+ */
+ String getCdcEngineName();
+
+ /**
+ * Registers the CDC event consumer.
+ *
+ * @param consumer The CDC event consumer to register.
+ */
+ void registerCdcEventConsumer(EventConsumer consumer);
+
+ void setContext(Context context);
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/CdcEngineFactory.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/CdcEngineFactory.java
new file mode 100644
index 000000000..531bd3f2b
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/CdcEngineFactory.java
@@ -0,0 +1,49 @@
+/*
+ * 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.connector.jdbc.DatabaseDialect;
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
+import org.apache.eventmesh.spi.EventMeshExtensionType;
+import org.apache.eventmesh.spi.EventMeshSPI;
+
+/**
+ * This interface defines the methods required to create a Change Data Capture (CDC) engine
+ */
+@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.JDBC_CDC_ENGINE)
+public interface CdcEngineFactory {
+
+ /**
+ * Determines whether the provided JDBC URL is compatible with the CDC engine
+ *
+ * @param url jdbc url, e.g. mysql: jdbc:mysql://localhost:3306/
+ * @return true if the JDBC URL is compatible with the CDC engine, false otherwise
+ */
+ boolean acceptJdbcProtocol(String url);
+
+ /**
+ * 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
+ */
+ CdcEngine createCdcEngine(SourceConfig config, DatabaseDialect databaseDialect);
+
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/RandomTaskSelectStrategy.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/RandomTaskSelectStrategy.java
new file mode 100644
index 000000000..4b05d41b3
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/RandomTaskSelectStrategy.java
@@ -0,0 +1,46 @@
+/*
+ * 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.connector.jdbc.source.EventMeshJdbcTask;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
+
+import java.util.List;
+import java.util.Random;
+
+public class RandomTaskSelectStrategy<Task extends EventMeshJdbcTask> implements TaskSelectStrategy<Task> {
+
+ private List<Task> cdcTasks;
+
+ public RandomTaskSelectStrategy(List<Task> cdcTasks) {
+ this.cdcTasks = cdcTasks;
+ }
+
+ /**
+ * Selects a JdbcTask for the specified TableId.
+ *
+ * @param tableId the TableId for which to select a JdbcTask
+ * @return the selected JdbcTask
+ */
+ @Override
+ public Task select(TableId tableId) {
+ Random random = new Random(System.currentTimeMillis());
+ int randomNum = random.nextInt(cdcTasks.size());
+ return cdcTasks.get(randomNum);
+ }
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/TaskSelectStrategy.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/TaskSelectStrategy.java
index 1de315b97..62e04c050 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/cdc/TaskSelectStrategy.java
@@ -15,32 +15,22 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source.dialect.cdc;
+
+import org.apache.eventmesh.connector.jdbc.source.EventMeshJdbcTask;
+import org.apache.eventmesh.connector.jdbc.table.catalog.TableId;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Represents a strategy for selecting a CdcTask for a given TableId.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+public interface TaskSelectStrategy<Task extends EventMeshJdbcTask> {
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Selects a CdcTask for the specified TableId.
+ *
+ * @param tableId the TableId for which to select a CdcTask
+ * @return the selected CdcTask
+ */
+ Task select(TableId tableId);
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotEngine.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotEngine.java
new file mode 100644
index 000000000..25cc19676
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotEngine.java
@@ -0,0 +1,45 @@
+/*
+ * 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.snapshot;
+
+import org.apache.eventmesh.connector.jdbc.JdbcContext;
+import org.apache.eventmesh.connector.jdbc.event.EventConsumer;
+import org.apache.eventmesh.connector.jdbc.source.Engine;
+
+/**
+ * The SnapshotEngine interface extends the Engine interface and represents an engine capable of performing snapshots.
+ *
+ * @param <Jc> The type of JdbcContext used for the snapshot.
+ */
+public interface SnapshotEngine<Jc extends JdbcContext> extends Engine, AutoCloseable {
+
+ /**
+ * Executes the snapshot operation and returns the result containing the snapshot offset.
+ *
+ * @return The SnapshotResult containing the snapshot offset.
+ */
+ SnapshotResult<Jc> execute();
+
+ /**
+ * Registers a SnapshotEventConsumer to receive snapshot events from the engine.
+ *
+ * @param consumer The SnapshotEventConsumer to register.
+ */
+ void registerSnapshotEventConsumer(EventConsumer consumer);
+}
+
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotEngineFactory.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotEngineFactory.java
new file mode 100644
index 000000000..33d756ba5
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotEngineFactory.java
@@ -0,0 +1,43 @@
+/*
+ * 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.snapshot;
+
+import org.apache.eventmesh.connector.jdbc.DatabaseDialect;
+import org.apache.eventmesh.connector.jdbc.JdbcContext;
+import org.apache.eventmesh.connector.jdbc.source.config.JdbcSourceConfig;
+import org.apache.eventmesh.spi.EventMeshExtensionType;
+import org.apache.eventmesh.spi.EventMeshSPI;
+
+/**
+ * The SnapshotEngineFactory interface represents a factory for creating snapshot engines. It provides a method to create a snapshot engine based on
+ * the given configuration and database dialect.
+ */
+@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.JDBC_SNAPSHOT_ENGINE)
+public interface SnapshotEngineFactory {
+
+ /**
+ * Creates a snapshot engine with the specified JDBC source configuration and database dialect.
+ *
+ * @param jdbcSourceConfig The JDBC source configuration.
+ * @param databaseDialect The database dialect.
+ * @return A snapshot engine that can perform snapshot operations.
+ */
+ SnapshotEngine<? extends JdbcContext> createSnapshotEngine(final JdbcSourceConfig jdbcSourceConfig,
+ final DatabaseDialect databaseDialect);
+}
+
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotResult.java
similarity index 57%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotResult.java
index 1de315b97..d36c85468 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotResult.java
@@ -15,32 +15,27 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source.dialect.snapshot;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+import org.apache.eventmesh.connector.jdbc.JdbcContext;
+
+import lombok.Getter;
+
+@Getter
+public class SnapshotResult<Jc extends JdbcContext> {
- public String getExtensionTypeName() {
- return extensionTypeName;
+ private final SnapshotResultStatus status;
+
+ private final Jc context;
+
+ public SnapshotResult(SnapshotResultStatus status, Jc jc) {
+ this.status = status;
+ this.context = jc;
}
+ public enum SnapshotResultStatus {
+ COMPLETED,
+ ABORTED,
+ SKIPPED
+ }
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotType.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotType.java
index 1de315b97..426835bb9 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/source/dialect/snapshot/SnapshotType.java
@@ -15,32 +15,17 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.source.dialect.snapshot;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+public enum SnapshotType {
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Every time the service starts, it reads the data of the tables that need to be processed.
+ */
+ INITIALIZATION,
+ /**
+ * Continue processing from the last handled position.
+ */
+ INCREASE //TODO Need to support next version
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Catalog.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Catalog.java
new file mode 100644
index 000000000..80f384429
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Catalog.java
@@ -0,0 +1,96 @@
+/*
+ * 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.table.catalog;
+
+import org.apache.eventmesh.connector.jdbc.exception.CatalogException;
+import org.apache.eventmesh.connector.jdbc.exception.DatabaseNotExistException;
+import org.apache.eventmesh.connector.jdbc.exception.TableNotExistException;
+
+import java.sql.SQLException;
+import java.util.List;
+
+/**
+ * Interacting with a catalog of databases and tables.
+ */
+public interface Catalog extends AutoCloseable {
+
+ /**
+ * Opens the catalog.
+ *
+ * @throws CatalogException if there is an error opening the catalog.
+ */
+ void open() throws CatalogException;
+
+ /**
+ * Gets the name of the default database.
+ *
+ * @return the name of the default database.
+ */
+ String getDefaultDatabase();
+
+ /**
+ * Checks if a database with the given name exists.
+ *
+ * @param databaseName the name of the database to check.
+ * @return true if the database exists, false otherwise.
+ * @throws CatalogException if there is an error checking for the database.
+ */
+ boolean databaseExists(String databaseName) throws CatalogException;
+
+ /**
+ * Gets a list of all databases in the catalog.
+ *
+ * @return a list of all databases in the catalog.
+ * @throws CatalogException if there is an error getting the list of databases.
+ */
+ List<String> listDatabases() throws CatalogException;
+
+ /**
+ * Gets a list of all tables in the given database.
+ *
+ * @param databaseName the name of the database to get the tables for.
+ * @return a list of all tables in the given database.
+ * @throws CatalogException if there is an error getting the list of tables.
+ * @throws DatabaseNotExistException if the database does not exist.
+ * @throws SQLException if there is an error accessing the database.
+ */
+ List<TableId> listTables(String databaseName) throws CatalogException, DatabaseNotExistException, SQLException;
+
+ /**
+ * Checks if a table with the given ID exists.
+ *
+ * @param tableId the ID of the table to check.
+ * @return true if the table exists, false otherwise.
+ * @throws CatalogException if there is an error checking for the table.
+ * @throws SQLException if there is an error accessing the database.
+ */
+ boolean tableExists(TableId tableId) throws CatalogException, SQLException;
+
+ /**
+ * Gets the table with the given ID.
+ *
+ * @param tableId the ID of the table to get.
+ * @return the table with the given ID.
+ * @throws CatalogException if there is an error getting the table.
+ * @throws TableNotExistException if the table does not exist.
+ * @throws SQLException if there is an error accessing the database.
+ */
+ CatalogTable getTable(TableId tableId) throws CatalogException, TableNotExistException, SQLException;
+
+ //TODO: support create table, drop table and update table
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/CatalogSchema.java
similarity index 57%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/CatalogSchema.java
index 1de315b97..2caae2a84 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/CatalogSchema.java
@@ -15,32 +15,33 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.table.catalog;
+
+import java.io.Serializable;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Represents catalog schema information, including its name and character set.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class CatalogSchema implements Serializable {
- OFFSETMGMT("offsetMgmt"),
- ;
+ /**
+ * The name of the catalog schema.
+ */
+ private String name;
- private final String extensionTypeName;
+ /**
+ * The character set used by the catalog schema.
+ */
+ private String characterSet;
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
+ public CatalogSchema(String name) {
+ this.name = name;
}
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
-
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/CatalogTable.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/CatalogTable.java
index 1de315b97..4aa2b8935 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/CatalogTable.java
@@ -15,32 +15,29 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.table.catalog;
+
+import java.io.Serializable;
+
+import lombok.Data;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Represents a catalog table.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+@Data
+public class CatalogTable implements Serializable {
+
+ private static final long serialVersionUID = -9159821671858779282L;
+
+ // The ID of the table.
+ private TableId tableId;
+
+ // The schema of the table.
+ private TableSchema tableSchema;
+
+ // A comment describing the table.
+ private String comment;
+
+ private Options options;
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Column.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Column.java
new file mode 100644
index 000000000..d31675af1
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Column.java
@@ -0,0 +1,102 @@
+/*
+ * 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.table.catalog;
+
+import org.apache.eventmesh.connector.jdbc.table.type.EventMeshDataType;
+
+import java.io.Serializable;
+import java.sql.JDBCType;
+import java.sql.Types;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Column of {@link TableSchema}.
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public abstract class Column implements Serializable {
+
+ /**
+ * Name of the column
+ */
+ protected String name;
+
+ /**
+ * Data type of the column
+ */
+ protected EventMeshDataType<?> dataType;
+
+ /**
+ * {@link Types JDBC type}
+ */
+ protected JDBCType jdbcType;
+
+ /**
+ * Length of the column
+ */
+ protected Integer columnLength;
+
+ /**
+ * Decimal point of the column
+ */
+ protected Integer decimal;
+
+ /**
+ * Indicates if the column can be null or not
+ */
+ protected boolean notNull;
+
+ /**
+ * Comment for the column
+ */
+ protected String comment;
+
+ /**
+ * Default value for the column
+ */
+ protected Object defaultValue;
+
+ protected String defaultValueExpression;
+
+ //protected int index;
+
+ /**
+ * Creates a new instance of the ColumnEditor.
+ *
+ * @return A new instance of the ColumnEditor.
+ */
+ public static ColumnEditor ofEditor() {
+ return new DefaultColumnEditorImpl();
+ }
+
+ public static ColumnEditor ofEditor(String name) {
+ return new DefaultColumnEditorImpl(name);
+ }
+
+ /**
+ * creates a clone of the Column
+ *
+ * @return clone of column
+ */
+ public abstract Column clone();
+
+}
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
new file mode 100644
index 000000000..df206c7cb
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/ColumnEditor.java
@@ -0,0 +1,149 @@
+/*
+ * 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.table.catalog;
+
+import org.apache.eventmesh.connector.jdbc.table.type.EventMeshDataType;
+
+import java.sql.JDBCType;
+import java.sql.Types;
+
+/**
+ * Interface for editing column properties.
+ */
+public interface ColumnEditor {
+
+ /**
+ * Sets the name of the column.
+ *
+ * @param name The name of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor withName(String name);
+
+ /**
+ * Sets the type of the column.
+ *
+ * @param typeName The type of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor withType(String typeName);
+
+ /**
+ * Sets the {@link Types JDBC type} of this column.
+ *
+ * @param jdbcType The JDBC type of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor withJdbcType(JDBCType jdbcType);
+
+ /**
+ * Sets the event mesh type of the column.
+ *
+ * @param eventMeshType The event mesh type of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor withEventMeshType(EventMeshDataType<?> eventMeshType);
+
+ /**
+ * Sets the character set name of the column.
+ *
+ * @param charsetName The character set name of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor charsetName(String charsetName);
+
+ /**
+ * Sets the character set name of the table associated with the column.
+ *
+ * @param charsetName The character set name of the table.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor charsetNameOfTable(String charsetName);
+
+ /**
+ * Sets the length of the column.
+ *
+ * @param length The length of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor length(int length);
+
+ /**
+ * Sets the scale of the column.
+ *
+ * @param scale The scale of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor scale(Integer scale);
+
+ /**
+ * Sets whether the column is optional.
+ *
+ * @param optional Whether the column is optional.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor optional(boolean optional);
+
+ /**
+ * Sets the comment of the column.
+ *
+ * @param comment The comment of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor comment(String comment);
+
+ /**
+ * Sets whether the column is auto-incremented.
+ *
+ * @param autoIncremented Whether the column is auto-incremented.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor autoIncremented(boolean autoIncremented);
+
+ /**
+ * Sets whether the column is generated.
+ *
+ * @param generated Whether the column is generated.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor generated(boolean generated);
+
+ /**
+ * Sets the default value expression of the column.
+ *
+ * @param defaultValueExpression The default value expression of the column.
+ * @return The ColumnEditor instance.
+ */
+ ColumnEditor defaultValueExpression(String defaultValueExpression);
+
+ /**
+ * Specifies whether the column should be marked as "NOT NULL."
+ *
+ * @param notNull True if the column should be marked as "NOT NULL," false otherwise.
+ * @return The updated ColumnEditor.
+ */
+ ColumnEditor notNull(boolean notNull);
+
+
+ /**
+ * Builds the Column object with the edited properties.
+ *
+ * @return The built Column object.
+ */
+ Column build();
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/DefaultColumn.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/DefaultColumn.java
new file mode 100644
index 000000000..8ba1d43b9
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/DefaultColumn.java
@@ -0,0 +1,51 @@
+/*
+ * 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.table.catalog;
+
+import org.apache.eventmesh.connector.jdbc.table.type.EventMeshDataType;
+
+import java.sql.JDBCType;
+
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+
+@EqualsAndHashCode(callSuper = true)
+@NoArgsConstructor
+public class DefaultColumn extends Column {
+
+ public DefaultColumn(String name, EventMeshDataType<?> dataType, JDBCType jdbcType, Integer columnLength, Integer decimal, boolean notNull,
+ String comment, Object defaultValue, String defaultValueExpression) {
+ super(name, dataType, jdbcType, columnLength, decimal, notNull, comment, defaultValue, defaultValueExpression);
+ }
+
+ public static DefaultColumn of(
+ String name, EventMeshDataType<?> dataType, JDBCType jdbcType, Integer columnLength, Integer decimal, boolean notNull,
+ String comment, Object defaultValue, String defaultValueExpression) {
+ return new DefaultColumn(name, dataType, jdbcType, columnLength, decimal, notNull, comment, defaultValue, defaultValueExpression);
+ }
+
+ /**
+ * creates a clone of the Column
+ *
+ * @return clone of column
+ */
+ @Override
+ public Column clone() {
+ return DefaultColumn.of(name, dataType, jdbcType, columnLength, decimal, notNull, comment, defaultValue, defaultValueExpression);
+ }
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/DefaultColumnEditorImpl.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/DefaultColumnEditorImpl.java
new file mode 100644
index 000000000..052aa1725
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/DefaultColumnEditorImpl.java
@@ -0,0 +1,258 @@
+/*
+ * 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.table.catalog;
+
+import org.apache.eventmesh.connector.jdbc.table.type.EventMeshDataType;
+
+import java.sql.JDBCType;
+
+public class DefaultColumnEditorImpl implements ColumnEditor {
+
+ /**
+ * Name of the column
+ */
+ private String name;
+
+ /**
+ * Data type of the column
+ */
+ private EventMeshDataType<?> dataType;
+
+ /**
+ * Length of the column
+ */
+ private Integer columnLength;
+
+ /**
+ * Decimal point of the column
+ */
+ private Integer decimal;
+
+ /**
+ * Indicates if the column can be null or not
+ */
+ private boolean notNull;
+
+ /**
+ * Comment for the column
+ */
+ private String comment;
+
+ /**
+ * Default value for the column
+ */
+ private Object defaultValue;
+
+ private String typeName;
+
+ private JDBCType jdbcType;
+
+ private String defaultValueExpression;
+
+ private String characterSet;
+
+ private boolean optional;
+
+ private boolean autoIncremented;
+
+ private boolean generated;
+
+ public DefaultColumnEditorImpl(String name) {
+ this.name = name;
+ }
+
+ public DefaultColumnEditorImpl() {
+ }
+
+ /**
+ * Sets the name of the column.
+ *
+ * @param name The name of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor withName(String name) {
+ this.name = name;
+ return this;
+ }
+
+ /**
+ * Sets the type of the column.
+ *
+ * @param typeName The type of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor withType(String typeName) {
+ this.typeName = typeName;
+ return this;
+ }
+
+ /**
+ * Sets the JDBC type of the column.
+ *
+ * @param jdbcType The JDBC type of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor withJdbcType(JDBCType jdbcType) {
+ this.jdbcType = jdbcType;
+ return this;
+ }
+
+ /**
+ * Sets the event mesh type of the column.
+ *
+ * @param eventMeshType The event mesh type of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor withEventMeshType(EventMeshDataType<?> eventMeshType) {
+ this.dataType = eventMeshType;
+ return this;
+ }
+
+ /**
+ * Sets the character set name of the column.
+ *
+ * @param charsetName The character set name of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor charsetName(String charsetName) {
+ this.characterSet = charsetName;
+ return this;
+ }
+
+ /**
+ * Sets the character set name of the table associated with the column.
+ *
+ * @param charsetName The character set name of the table.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor charsetNameOfTable(String charsetName) {
+ return this;
+ }
+
+ /**
+ * Sets the length of the column.
+ *
+ * @param length The length of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor length(int length) {
+ this.columnLength = length;
+ return this;
+ }
+
+ /**
+ * Sets the scale of the column.
+ *
+ * @param scale The scale of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor scale(Integer scale) {
+ this.decimal = scale;
+ return this;
+ }
+
+ /**
+ * Sets whether the column is optional.
+ *
+ * @param optional Whether the column is optional.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor optional(boolean optional) {
+ this.optional = optional;
+ return this;
+ }
+
+ /**
+ * Sets the comment of the column.
+ *
+ * @param comment The comment of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor comment(String comment) {
+ this.comment = comment;
+ return this;
+ }
+
+ /**
+ * Sets whether the column is auto-incremented.
+ *
+ * @param autoIncremented Whether the column is auto-incremented.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor autoIncremented(boolean autoIncremented) {
+ this.autoIncremented = autoIncremented;
+ return this;
+ }
+
+ /**
+ * Sets whether the column is generated.
+ *
+ * @param generated Whether the column is generated.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor generated(boolean generated) {
+ this.generated = generated;
+ return this;
+ }
+
+ /**
+ * Sets the default value expression of the column.
+ *
+ * @param defaultValueExpression The default value expression of the column.
+ * @return The ColumnEditor instance.
+ */
+ @Override
+ public ColumnEditor defaultValueExpression(String defaultValueExpression) {
+ this.defaultValueExpression = defaultValueExpression;
+ return this;
+ }
+
+ /**
+ * Specifies whether the column should be marked as "NOT NULL."
+ *
+ * @param notNull True if the column should be marked as "NOT NULL," false otherwise.
+ * @return The updated ColumnEditor.
+ */
+ @Override
+ public ColumnEditor notNull(boolean notNull) {
+ this.notNull = notNull;
+ return this;
+ }
+
+ /**
+ * Builds the Column object with the edited properties.
+ *
+ * @return The built Column object.
+ */
+ @Override
+ public Column build() {
+ return DefaultColumn.of(name, dataType, jdbcType, columnLength, decimal, notNull, comment, defaultValue, defaultValueExpression);
+ }
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Options.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Options.java
index 1de315b97..de2429695 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Options.java
@@ -15,32 +15,10 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.table.catalog;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+import java.util.HashMap;
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+public class Options extends HashMap<String, Object> {
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/PrimaryKey.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/PrimaryKey.java
new file mode 100644
index 000000000..521733943
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/PrimaryKey.java
@@ -0,0 +1,59 @@
+/*
+ * 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.table.catalog;
+
+import java.util.List;
+
+public class PrimaryKey extends UniqueKey {
+
+
+ public PrimaryKey(String name, List<String> columnNames, String comment) {
+ super(name, columnNames, comment);
+ }
+
+ public PrimaryKey(String name, List<String> columnNames) {
+ super(name, columnNames);
+ }
+
+ public PrimaryKey(List<String> columnNames) {
+ super(columnNames);
+ }
+
+ public PrimaryKey(List<String> columnNames, String comment) {
+ super(null, columnNames, comment);
+ }
+
+ /**
+ * Creates a new PrimaryKey instance with the given name and column names.
+ *
+ * @param columnNames The list of column names that make up the primary key.
+ * @return A new PrimaryKey instance.
+ */
+ public static PrimaryKey of(List<String> columnNames) {
+ return new PrimaryKey(columnNames);
+ }
+
+ /**
+ * Creates a copy of this PrimaryKey instance.
+ *
+ * @return A new PrimaryKey instance with the same name and column names.
+ */
+ public PrimaryKey copy() {
+ return new PrimaryKey(getName(), getColumnNames(), getComment());
+ }
+}
\ No newline at end of file
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Table.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Table.java
index 1de315b97..4e9077191 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/Table.java
@@ -15,32 +15,18 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.table.catalog;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+import java.util.List;
+
+public class Table {
+
+ private String name;
+
+ private PrimaryKey primaryKey;
+
+ private List<UniqueKey> uniqueKeys;
+
+ private String comment;
}
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
new file mode 100644
index 000000000..acecc833e
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableId.java
@@ -0,0 +1,156 @@
+/*
+ * 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.table.catalog;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Represents a table identifier with catalog name, schema name and table name.
+ */
+@Getter
+@Setter
+public class TableId implements Serializable {
+
+ /**
+ * The default mapper that converts a TableId to its string representation.
+ */
+ public static final TableIdToStringMapper DEFAULT_TABLEIDTOSTRINGMAPPER = new DefaultTableIdToStringMapper();
+
+ private String catalogName;
+
+ private String schemaName;
+
+ private String tableName;
+
+ private String id;
+
+ public TableId() {
+ }
+
+ /**
+ * Constructs a TableId instance without a TableIdToStringMapper.
+ *
+ * @param catalogName the catalog name of the table
+ * @param schemaName the schema name of the table
+ * @param tableName the name of the table
+ */
+ public TableId(String catalogName, String schemaName, String tableName) {
+ this(catalogName, schemaName, tableName, null);
+ }
+
+ /**
+ * Constructs a TableId instance without a TableIdToStringMapper.
+ *
+ * @param catalogName the catalog name of the table
+ */
+ public TableId(String catalogName) {
+ this(catalogName, null, null, null);
+ }
+
+ /**
+ * Constructs a TableId instance with a TableIdToStringMapper. If the mapper is null, the default mapper will be used.
+ *
+ * @param catalogName the catalog name of the table
+ * @param schemaName the schema name of the table
+ * @param tableName the name of the table
+ * @param mapper the mapper that converts a TableId to its string representation
+ */
+ public TableId(String catalogName, String schemaName, String tableName, TableIdToStringMapper mapper) {
+ this.catalogName = catalogName;
+ this.schemaName = schemaName;
+ this.tableName = tableName;
+ this.id = mapper == null ? DEFAULT_TABLEIDTOSTRINGMAPPER.toString(this) : mapper.toString(this);
+
+ }
+
+ @Override
+ public String toString() {
+ return id == null ? DEFAULT_TABLEIDTOSTRINGMAPPER.toString(this) : id;
+ }
+
+ /**
+ * Returns the string representation of the TableId, which is the same as calling toString().
+ *
+ * @return the string representation of the TableId
+ */
+ public String tablePath() {
+ return id;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof TableId)) {
+ return false;
+ }
+ TableId tableId = (TableId) o;
+ return Objects.equals(getCatalogName(), tableId.getCatalogName()) && Objects.equals(getSchemaName(), tableId.getSchemaName())
+ && Objects.equals(getTableName(), tableId.getTableName()) && Objects.equals(getId(), tableId.getId());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(getCatalogName(), getSchemaName(), getTableName(), getId());
+ }
+
+ /**
+ * A functional interface that converts a TableId to its string representation.
+ */
+ @FunctionalInterface
+ public interface TableIdToStringMapper {
+
+ String toString(TableId tableId);
+ }
+
+ /**
+ * Returns the string representation of a TableId. If catalog or schema is null or empty, they will be excluded from the string.
+ *
+ * @param catalog the catalog name of the table
+ * @param schema the schema name of the table
+ * @param table the name of the table
+ * @return the string representation of the TableId
+ */
+ private static String tableId(String catalog, String schema, String table) {
+ if (catalog == null || catalog.length() == 0) {
+ if (schema == null || schema.length() == 0) {
+ return table;
+ }
+ return schema + "." + table;
+ }
+ if (schema == null || schema.length() == 0) {
+ return catalog + "." + table;
+ }
+ return catalog + "." + schema + "." + table;
+ }
+
+ /**
+ * The default mapper that converts a TableId to its string representation.
+ */
+ private static class DefaultTableIdToStringMapper implements TableIdToStringMapper {
+
+ public String toString(TableId tableId) {
+ return tableId(tableId.getCatalogName(), tableId.getSchemaName(), tableId.getTableName());
+ }
+ }
+}
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
new file mode 100644
index 000000000..3d2c0446f
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/TableSchema.java
@@ -0,0 +1,112 @@
+/*
+ * 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.table.catalog;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+public class TableSchema implements Serializable {
+
+ private String name;
+
+ /**
+ * A map of column names to their respective column objects.
+ */
+ private Map<String, ? extends Column> columnMap;
+
+ /**
+ * A list of columns in the table.
+ */
+ private List<? extends Column> columns;
+
+ /**
+ * The primary key of the table.
+ */
+ private PrimaryKey primaryKey;
+
+ private List<UniqueKey> uniqueKeys;
+
+ private String comment;
+
+ public TableSchema(String name) {
+ this.name = name;
+ }
+
+ public static TableSchemaBuilder newTableSchemaBuilder() {
+ return new TableSchemaBuilder();
+ }
+
+ public static class TableSchemaBuilder {
+
+ private String name;
+ private Map<String, Column> columnMap;
+ private List<Column> columns;
+ private PrimaryKey primaryKey;
+ private List<UniqueKey> uniqueKeys;
+ private String comment;
+
+ public TableSchemaBuilder() {
+
+ }
+
+ public TableSchemaBuilder withName(String name) {
+ this.name = name;
+ return this;
+ }
+
+ public TableSchemaBuilder withColumns(Map<String, Column> columnMap) {
+ this.columnMap = columnMap;
+ return this;
+ }
+
+ public TableSchemaBuilder withColumns(List<Column> columns) {
+ this.columns = columns;
+ return this;
+ }
+
+ public TableSchemaBuilder withPrimaryKey(PrimaryKey primaryKey) {
+ this.primaryKey = primaryKey;
+ return this;
+ }
+
+ public TableSchemaBuilder withUniqueKeys(List<UniqueKey> uniqueKeys) {
+ this.uniqueKeys = uniqueKeys;
+ return this;
+ }
+
+ public TableSchemaBuilder withComment(String comment) {
+ this.comment = comment;
+ return this;
+ }
+
+ public TableSchema build() {
+ return new TableSchema(name, columnMap, columns, primaryKey, uniqueKeys, comment);
+ }
+
+ }
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/UniqueKey.java
similarity index 50%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/UniqueKey.java
index 1de315b97..c589ae381 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/catalog/UniqueKey.java
@@ -15,32 +15,42 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.table.catalog;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
+import java.io.Serializable;
+import java.util.List;
+
+import lombok.Getter;
+import lombok.Setter;
+
+
+@Setter
+@Getter
+public class UniqueKey implements Serializable {
+
+ private String name;
+
+ private final List<String> columnNames;
+
+ private String comment;
+
+ public UniqueKey(String name, List<String> columnNames, String comment) {
+ this.name = name;
+ this.columnNames = columnNames;
+ this.comment = comment;
+ }
+
+ public UniqueKey(String name, List<String> columnNames) {
+ this.name = name;
+ this.columnNames = columnNames;
}
- public String getExtensionTypeName() {
- return extensionTypeName;
+ public UniqueKey(List<String> columnNames) {
+ this.columnNames = columnNames;
}
+ public UniqueKey copy() {
+ return new UniqueKey(name, columnNames, comment);
+ }
}
+
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/EventMeshDataType.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/EventMeshDataType.java
index 1de315b97..d2a2df1be 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/EventMeshDataType.java
@@ -15,32 +15,24 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.table.type;
/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
+ * Defines Event Mesh data type with methods to get the type class and SQL type of the data.
*/
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
+public interface EventMeshDataType<T> {
- OFFSETMGMT("offsetMgmt"),
- ;
-
- private final String extensionTypeName;
-
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
-
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ /**
+ * Gets the type class of the data.
+ *
+ * @return the type class of the data.
+ */
+ Class<T> getTypeClass();
+ /**
+ * Gets the SQL type of the data.
+ *
+ * @return the SQL type of the data.
+ */
+ SQLType getSQLType();
}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/Pair.java
similarity index 56%
copy from eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
copy to eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/Pair.java
index 1de315b97..e5f0dd72a 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/Pair.java
@@ -15,32 +15,21 @@
* limitations under the License.
*/
-package org.apache.eventmesh.spi;
+package org.apache.eventmesh.connector.jdbc.table.type;
-/**
- * An Extension can be defined by extensionTypeName and extensionInstanceName
- */
-public enum EventMeshExtensionType {
- UNKNOWN("unknown"),
- CONNECTOR("connector"),
- STORAGE("storage"),
- REGISTRY("registry"),
- SECURITY("security"),
- PROTOCOL("protocol"),
- METRICS("metrics"),
- TRACE("trace"),
-
- OFFSETMGMT("offsetMgmt"),
- ;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
- private final String extensionTypeName;
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+public class Pair<Left, Right> {
- EventMeshExtensionType(String extensionTypeName) {
- this.extensionTypeName = extensionTypeName;
- }
+ private Left left;
- public String getExtensionTypeName() {
- return extensionTypeName;
- }
+ private Right right;
}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/RowHandleMode.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/RowHandleMode.java
new file mode 100644
index 000000000..ed428f132
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/RowHandleMode.java
@@ -0,0 +1,86 @@
+/*
+ * 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.table.type;
+
+/**
+ * An enum representing the different modes in which a row can be handled.
+ */
+public enum RowHandleMode {
+ INSERT("+I", (byte) 1),
+ UPDATE_BEFORE("-UB", (byte) 2),
+ UPDATE_AFTER("+UA", (byte) 3),
+ DELETE("-D", (byte) 1),
+ ;
+
+ private final String shortCut;
+
+ private final byte value;
+
+ /**
+ * Constructor for RowHandleMode.
+ *
+ * @param shortCut a string representing the shorthand for the row handle mode.
+ * @param value a byte representing the value of the row handle mode.
+ */
+ RowHandleMode(String shortCut, byte value) {
+ this.shortCut = shortCut;
+ this.value = value;
+ }
+
+ /**
+ * Returns the shorthand for the row handle mode.
+ *
+ * @return a string representing the shorthand for the row handle mode.
+ */
+ public String toShortCut() {
+ return shortCut;
+ }
+
+ /**
+ * Returns the value of the row handle mode.
+ *
+ * @return a byte representing the value of the row handle mode.
+ */
+ public byte toValue() {
+ return value;
+ }
+
+ /**
+ * Returns the row handle mode corresponding to the given byte value.
+ *
+ * @param value a byte representing the value of the row handle mode.
+ * @return the row handle mode corresponding to the given byte value.
+ * @throws UnsupportedOperationException if the byte value is not supported.
+ */
+ public static RowHandleMode fromByteValue(byte value) {
+ switch (value) {
+ case 0:
+ return INSERT;
+ case 1:
+ return UPDATE_BEFORE;
+ case 2:
+ return UPDATE_AFTER;
+ case 3:
+ return DELETE;
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported byte value '" + value + "' for row handle mode.");
+ }
+ }
+}
+
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/SQLType.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/SQLType.java
new file mode 100644
index 000000000..203ceb84b
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/table/type/SQLType.java
@@ -0,0 +1,96 @@
+/*
+ * 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.table.type;
+
+/**
+ * see {@link java.sql.SQLType}
+ */
+public enum SQLType {
+
+ /**
+ * Identifies the generic SQL type {@code TINYINT}.
+ */
+ TINYINT,
+ /**
+ * Identifies the generic SQL type {@code SMALLINT}.
+ */
+ SMALLINT,
+ /**
+ * Identifies the generic SQL type {@code INTEGER}.
+ */
+ INTEGER,
+ /**
+ * Identifies the generic SQL type {@code BIGINT}.
+ */
+ BIGINT,
+ /**
+ * Identifies the generic SQL type {@code FLOAT}.
+ */
+ FLOAT,
+
+ /**
+ * Identifies the generic SQL type {@code DOUBLE}.
+ */
+ DOUBLE,
+
+ /**
+ * Identifies the generic SQL type {@code DECIMAL}.
+ */
+ DECIMAL,
+
+ /**
+ * Identifies the generic SQL type {@code DATE}.
+ */
+ DATE,
+ /**
+ * Identifies the generic SQL type {@code TIME}.
+ */
+ TIME,
+ /**
+ * Identifies the generic SQL type {@code TIMESTAMP}.
+ */
+ TIMESTAMP,
+ /**
+ * Identifies the generic SQL type {@code BINARY}.
+ */
+ BINARY,
+
+ /**
+ * Identifies the generic SQL value {@code NULL}.
+ */
+ NULL,
+
+ /**
+ * Identifies the generic SQL type {@code ARRAY}.
+ */
+ ARRAY,
+
+ /**
+ * Identifies the generic SQL type {@code BOOLEAN}.
+ */
+ BOOLEAN,
+
+ /**
+ * EventMesh generic SQL type
+ */
+ ROW,
+
+ MAP,
+
+ STRING
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/utils/JdbcStringUtils.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/utils/JdbcStringUtils.java
new file mode 100644
index 000000000..d74017737
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/utils/JdbcStringUtils.java
@@ -0,0 +1,50 @@
+/*
+ * 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.utils;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class JdbcStringUtils {
+
+ public static boolean isWrapped(String possiblyWrapped) {
+ if (possiblyWrapped.length() < 2) {
+ return false;
+ }
+ if (possiblyWrapped.startsWith("`") && possiblyWrapped.endsWith("`")) {
+ return true;
+ }
+ if (possiblyWrapped.startsWith("'") && possiblyWrapped.endsWith("'")) {
+ return true;
+ }
+ if (possiblyWrapped.startsWith("\"") && possiblyWrapped.endsWith("\"")) {
+ return true;
+ }
+ return false;
+ }
+
+ public static boolean isWrapped(char c) {
+ return c == '\'' || c == '"' || c == '`';
+ }
+
+ public static String withoutWrapper(String possiblyWrapped) {
+ return isWrapped(possiblyWrapped) ? possiblyWrapped.substring(1, possiblyWrapped.length() - 1) : possiblyWrapped;
+ }
+
+
+}
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.DatabaseDialectFactory b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.DatabaseDialectFactory
new file mode 100644
index 000000000..8524b42ac
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.DatabaseDialectFactory
@@ -0,0 +1,16 @@
+# 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.
+
+mysql=org.apache.eventmesh.connector.jdbc.source.dialect.mysql.MysqlDatabaseDialectFactory
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
new file mode 100644
index 000000000..06eb1b99e
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.cdc.CdcEngineFactory
@@ -0,0 +1,16 @@
+# 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.
+
+mysql=org.apache.eventmesh.connector.jdbc.source.dialect.cdc.mysql.MysqlCdcEngineFactory
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.SnapshotEngineFactory b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.SnapshotEngineFactory
new file mode 100644
index 000000000..39075175e
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.SnapshotEngineFactory
@@ -0,0 +1,16 @@
+# 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.
+
+mysql=org.apache.eventmesh.connector.jdbc.source.dialect.snapshot.mysql.MysqlSnapshotEngineFactory
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/log4j2.xml b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/log4j2.xml
new file mode 100644
index 000000000..cd699adad
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/log4j2.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<Configuration status="debug" monitorInterval="60">
+ <Appenders>
+ <Console name="console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p [%t] %c{1}(%F:%L) - %m%n"/>
+ </Console>
+ </Appenders>
+
+ <Loggers>
+ <AsyncRoot level="DEBUG" includeLocation="true">
+ <AppenderRef ref="console"/>
+ </AsyncRoot>
+ </Loggers>
+</Configuration>
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/server-config.yml b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/server-config.yml
new file mode 100644
index 000000000..5f66dd0f6
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/server-config.yml
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+sourceEnable: true
+sinkEnable: true
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/source-config.yml b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/source-config.yml
new file mode 100644
index 000000000..cc841fca7
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/main/resources/source-config.yml
@@ -0,0 +1,60 @@
+#
+# 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.
+#
+
+pubSubConfig:
+ meshAddress: 127.0.0.1:10001
+ subject: TopicTest
+ idc: FT
+ env: PRD
+ group: rocketmqSource
+ appId: 5032
+ userName: jdbcSourceUser
+ passWord: jdbcPassWord
+sourceConnectorConfig:
+ maxTask: 10 #max task number
+ batchMaxRows: 100
+ skipSnapshot: false
+ snapshotMaxThreads: 10
+ snapshotSchema: true
+ snapshotData: true
+ snapshotFetchSize: 100
+ databaseType: mysql
+ databaseIncludeList:
+ - mxsm
+ databaseExcludeList:
+ tableIncludeList:
+ tableExcludeList:
+ jdbcConfig:
+ hostname: localhost
+ port: 3306
+ user: root
+ password: Mxsm22##
+ initialStatements:
+ connectTimeout: 10
+ mysqlConfig:
+ serverId: 123
+ keepAlive: true
+ keepAliveInterval: 6000
+offsetStorageConfig:
+ offsetStorageType: nacos
+ offsetStorageAddr: 127.0.0.1:8848
+ extensions: {
+ #same with topic
+ dataId: TopicTest,
+ #same with group
+ group: rocketmqSource
+ }
\ No newline at end of file
diff --git a/eventmesh-connectors/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/utils/JdbcStringUtilsTest.java b/eventmesh-connectors/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/utils/JdbcStringUtilsTest.java
new file mode 100644
index 000000000..e69f01753
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/utils/JdbcStringUtilsTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.utils;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+public class JdbcStringUtilsTest {
+
+ @Test
+ public void testIsWrapped() {
+ assertTrue(JdbcStringUtils.isWrapped("`Hello`"));
+ assertTrue(JdbcStringUtils.isWrapped("'World'"));
+ assertTrue(JdbcStringUtils.isWrapped("\"Java\""));
+ assertFalse(JdbcStringUtils.isWrapped("NotWrapped"));
+ assertFalse(JdbcStringUtils.isWrapped("`NotClosed"));
+ assertFalse(JdbcStringUtils.isWrapped("NotOpened`"));
+ }
+
+ @Test
+ public void testWithoutWrapper() {
+ assertEquals("Hello", JdbcStringUtils.withoutWrapper("`Hello`"));
+ assertEquals("World", JdbcStringUtils.withoutWrapper("'World'"));
+ assertEquals("Java", JdbcStringUtils.withoutWrapper("\"Java\""));
+ assertEquals("NotWrapped", JdbcStringUtils.withoutWrapper("NotWrapped"));
+ assertEquals("`NotClosed", JdbcStringUtils.withoutWrapper("`NotClosed"));
+ assertEquals("NotOpened`", JdbcStringUtils.withoutWrapper("NotOpened`"));
+ }
+
+ @Test
+ public void testIsWrappedWithChar() {
+ assertTrue(JdbcStringUtils.isWrapped('`'));
+ assertTrue(JdbcStringUtils.isWrapped('\''));
+ assertTrue(JdbcStringUtils.isWrapped('\"'));
+ assertFalse(JdbcStringUtils.isWrapped('A'));
+ }
+}
diff --git a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java b/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
index 1de315b97..49daf4e31 100644
--- a/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
+++ b/eventmesh-spi/src/main/java/org/apache/eventmesh/spi/EventMeshExtensionType.java
@@ -29,7 +29,9 @@ public enum EventMeshExtensionType {
PROTOCOL("protocol"),
METRICS("metrics"),
TRACE("trace"),
-
+ JDBC_CDC_ENGINE("jdbc_cdc_engine"),
+ JDBC_SNAPSHOT_ENGINE("jdbc_snapshot_engine"),
+ JDBC_DATABASE_DIALECT("jdbc_database_dialect"),
OFFSETMGMT("offsetMgmt"),
;
diff --git a/settings.gradle b/settings.gradle
index 4df089780..438721c5f 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -38,6 +38,7 @@ include 'eventmesh-connectors:eventmesh-connector-kafka'
include 'eventmesh-connectors:eventmesh-connector-s3'
include 'eventmesh-connectors:eventmesh-connector-pravega'
+include 'eventmesh-connectors:eventmesh-connector-jdbc'
include 'eventmesh-storage-plugin:eventmesh-storage-api'
include 'eventmesh-storage-plugin:eventmesh-storage-standalone'
include 'eventmesh-storage-plugin:eventmesh-storage-kafka'
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org