This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new d13fe5b Debezium: integrate KafkaConnectSource with
MySqlConnectorTask (#2791)
d13fe5b is described below
commit d13fe5bc287d2bc082da1682703d01dc92140a3b
Author: Jia Zhai <jiaz...@users.noreply.github.com>
AuthorDate: Mon Oct 29 14:41:36 2018 +0800
Debezium: integrate KafkaConnectSource with MySqlConnectorTask (#2791)
### Motivation
This change is to integrate KafkaConnectSource with MySqlConnectorTask.
Steps to run this is like this:
1, use image in debezium tutorial, start mysql server
```docker run -it --rm --name mysql -p 3306:3306 -e
MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw
debezium/example-mysql:0.8```
2, start pulsar standalone
```bin/pulsar standalone ```
3, start consume data from connect source target topic, After step 4 is
start, it should get data.
```bin/pulsar-client consume -s "my-sub-name" kafka-connect-topic -n 0```
4, start source connect, It should produce data into Pulsar topics.
```bin/pulsar-admin source localrun --tenant public --namespace default
--name kafka --destination-topic-name kafka-connect-topic --sourceConfigFile
debezium-mysql-source-config.yaml --archive
connectors/pulsar-io-kafka-connect-adaptor-2.2.0-SNAPSHOT.nar
--brokerServiceUrl pulsar://127.0.0.1:6650```
---
distribution/io/src/assemble/io.xml | 6 +++
pom.xml | 2 +-
pulsar-io/debezium/pom.xml | 8 ++--
.../pulsar/io/debezium/PulsarDatabaseHistory.java | 2 +-
pulsar-io/kafka-connect-adaptor/pom.xml | 22 ++++++++++
.../io/kafka/connect/KafkaConnectSource.java | 27 +++++++++---
.../resources/META-INF/services/pulsar-io.yaml | 22 ++++++++++
.../resources/debezium-mysql-source-config.yaml | 51 ++++++++++++++++++++++
.../io/kafka/connect/KafkaConnectSourceTest.java | 6 +--
9 files changed, 131 insertions(+), 15 deletions(-)
diff --git a/distribution/io/src/assemble/io.xml
b/distribution/io/src/assemble/io.xml
index c527fc7..1a190a4 100644
--- a/distribution/io/src/assemble/io.xml
+++ b/distribution/io/src/assemble/io.xml
@@ -93,5 +93,11 @@
<outputDirectory>connectors</outputDirectory>
<fileMode>644</fileMode>
</file>
+
+ <file>
+
<source>${basedir}/../../pulsar-io/kafka-connect-adaptor/target/pulsar-io-kafka-connect-adaptor-${project.version}.nar</source>
+ <outputDirectory>connectors</outputDirectory>
+ <fileMode>644</fileMode>
+ </file>
</files>
</assembly>
diff --git a/pom.xml b/pom.xml
index eac83c0..ef870ab 100644
--- a/pom.xml
+++ b/pom.xml
@@ -178,7 +178,7 @@ flexible messaging model and an intuitive client
API.</description>
<presto.version>0.206</presto.version>
<flink.version>1.6.0</flink.version>
<scala.binary.version>2.11</scala.binary.version>
- <debezium-core.version>0.8.2</debezium-core.version>
+ <debezium.version>0.8.2</debezium.version>
<!-- test dependencies -->
<arquillian-cube.version>1.15.1</arquillian-cube.version>
diff --git a/pulsar-io/debezium/pom.xml b/pulsar-io/debezium/pom.xml
index fb79196..3a178ab 100644
--- a/pulsar-io/debezium/pom.xml
+++ b/pulsar-io/debezium/pom.xml
@@ -41,13 +41,13 @@
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-core</artifactId>
- <version>${debezium-core.version}</version>
+ <version>${debezium.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.pulsar</groupId>
- <artifactId>pulsar-io-kafka-connect-adaptor</artifactId>
- <version>${project.version}</version>
+ <groupId>io.debezium</groupId>
+ <artifactId>debezium-connector-mysql</artifactId>
+ <version>${debezium.version}</version>
</dependency>
<dependency>
diff --git
a/pulsar-io/debezium/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java
b/pulsar-io/debezium/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java
index bc97fc6..820c5a4 100644
---
a/pulsar-io/debezium/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java
+++
b/pulsar-io/debezium/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java
@@ -61,7 +61,7 @@ public final class PulsarDatabaseHistory extends
AbstractDatabaseHistory {
.withValidation(Field::isRequired);
public static final Field SERVICE_URL =
Field.create(CONFIGURATION_FIELD_PREFIX_STRING + "pulsar.service.url")
- .withDisplayName("Kafka broker addresses")
+ .withDisplayName("Pulsar broker addresses")
.withType(Type.STRING)
.withWidth(Width.LONG)
.withImportance(Importance.HIGH)
diff --git a/pulsar-io/kafka-connect-adaptor/pom.xml
b/pulsar-io/kafka-connect-adaptor/pom.xml
index e0f0574..b2d0eea 100644
--- a/pulsar-io/kafka-connect-adaptor/pom.xml
+++ b/pulsar-io/kafka-connect-adaptor/pom.xml
@@ -39,6 +39,12 @@
</dependency>
<dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-debezium</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_${scala.binary.version}</artifactId>
<version>${kafka-client.version}</version>
@@ -52,6 +58,12 @@
<dependency>
<groupId>org.apache.kafka</groupId>
+ <artifactId>connect-json</artifactId>
+ <version>${kafka-client.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
<artifactId>connect-api</artifactId>
<version>${kafka-client.version}</version>
</dependency>
@@ -102,4 +114,14 @@
</dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-nar-maven-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
diff --git
a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
index 61635f8..5aac5dd 100644
---
a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
+++
b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
@@ -23,6 +23,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -74,21 +75,25 @@ public class KafkaConnectSource implements Source<byte[]> {
});
// get the source class name from config and create source task from
reflection
- sourceTask = ((Class<? extends
SourceTask>)config.get(TaskConfig.TASK_CLASS_CONFIG))
+ sourceTask = ((Class<? extends
SourceTask>)Class.forName(stringConfig.get(TaskConfig.TASK_CLASS_CONFIG)))
.asSubclass(SourceTask.class)
.getDeclaredConstructor()
.newInstance();
+
// initialize the key and value converter
- keyConverter = ((Class<? extends
Converter>)config.get(PulsarKafkaWorkerConfig.KEY_CONVERTER_CLASS_CONFIG))
+ keyConverter = ((Class<? extends
Converter>)Class.forName(stringConfig.get(PulsarKafkaWorkerConfig.KEY_CONVERTER_CLASS_CONFIG)))
.asSubclass(Converter.class)
.getDeclaredConstructor()
.newInstance();
- valueConverter = ((Class<? extends
Converter>)config.get(PulsarKafkaWorkerConfig.VALUE_CONVERTER_CLASS_CONFIG))
+ valueConverter = ((Class<? extends
Converter>)Class.forName(stringConfig.get(PulsarKafkaWorkerConfig.VALUE_CONVERTER_CLASS_CONFIG)))
.asSubclass(Converter.class)
.getDeclaredConstructor()
.newInstance();
+ keyConverter.configure(config, true);
+ valueConverter.configure(config, false);
+
offsetStore = new PulsarOffsetBackingStore();
offsetStore.configure(new PulsarKafkaWorkerConfig(stringConfig));
offsetStore.start();
@@ -117,7 +122,12 @@ public class KafkaConnectSource implements Source<byte[]> {
while (true) {
if (currentBatch == null) {
flushFuture = new CompletableFuture<>();
- currentBatch = sourceTask.poll().iterator();
+ List<SourceRecord> recordList = sourceTask.poll();
+ if (recordList == null) {
+ Thread.sleep(1000);
+ continue;
+ }
+ currentBatch = recordList.iterator();
}
if (currentBatch.hasNext()) {
return processSourceRecord(currentBatch.next());
@@ -126,7 +136,7 @@ public class KafkaConnectSource implements Source<byte[]> {
synchronized (this) {
hasOutstandingRecords = !outstandingRecords.isEmpty();
}
- if (hasOutstandingRecords) {
+ if (!hasOutstandingRecords) {
// there is no records any more, then waiting for the
batch to complete writing
// to sink and the offsets are committed as well
flushFuture.get();
@@ -161,7 +171,7 @@ public class KafkaConnectSource implements Source<byte[]> {
@Override
public Optional<Long> getEventTime() {
- return Optional.of(srcRecord.timestamp());
+ return Optional.ofNullable(srcRecord.timestamp());
}
@Override
@@ -202,6 +212,11 @@ public class KafkaConnectSource implements Source<byte[]> {
flushFuture.completeExceptionally(new Exception("Sink
Error"));
}
}
+
+ @Override
+ public Optional<String> getDestinationTopic() {
+ return Optional.of(srcRecord.topic());
+ }
};
}
diff --git
a/pulsar-io/kafka-connect-adaptor/src/main/resources/META-INF/services/pulsar-io.yaml
b/pulsar-io/kafka-connect-adaptor/src/main/resources/META-INF/services/pulsar-io.yaml
new file mode 100644
index 0000000..4887063
--- /dev/null
+++
b/pulsar-io/kafka-connect-adaptor/src/main/resources/META-INF/services/pulsar-io.yaml
@@ -0,0 +1,22 @@
+#
+# 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.
+#
+
+name: kafka-connect-adaptor
+description: Kafka source connect adaptor
+sourceClass: org.apache.pulsar.io.kafka.connect.KafkaConnectSource
diff --git
a/pulsar-io/kafka-connect-adaptor/src/main/resources/debezium-mysql-source-config.yaml
b/pulsar-io/kafka-connect-adaptor/src/main/resources/debezium-mysql-source-config.yaml
new file mode 100644
index 0000000..2e288cb
--- /dev/null
+++
b/pulsar-io/kafka-connect-adaptor/src/main/resources/debezium-mysql-source-config.yaml
@@ -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.
+#
+
+tenant: "test"
+namespace: "test-namespace"
+name: "debezium-kafka-source"
+
+##autoAck: true
+parallelism: 1
+
+configs:
+ ## sourceTask
+ task.class: "io.debezium.connector.mysql.MySqlConnectorTask"
+
+ ## config for mysql, docker image: debezium/example-mysql:0.8
+ database.hostname: "localhost"
+ database.port: "3306"
+ database.user: "debezium"
+ database.password: "dbz"
+ database.server.id: "184054"
+ database.server.name: "dbserver1"
+ database.whitelist: "inventory"
+
+ database.history: "org.apache.pulsar.io.debezium.PulsarDatabaseHistory"
+ database.history.pulsar.topic: "history-topic"
+ database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ ## KEY_CONVERTER_CLASS_CONFIG, VALUE_CONVERTER_CLASS_CONFIG
+ key.converter: "org.apache.kafka.connect.json.JsonConverter"
+ value.converter: "org.apache.kafka.connect.json.JsonConverter"
+ ## PULSAR_SERVICE_URL_CONFIG
+ pulsar.service.url: "pulsar://127.0.0.1:6650"
+ ## OFFSET_STORAGE_TOPIC_CONFIG
+ offset.storage.topic: "offset-topic"
+
+
diff --git
a/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
b/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
index b82daab..d0f0a63 100644
---
a/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
+++
b/pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java
@@ -70,9 +70,9 @@ public class KafkaConnectSourceTest extends
ProducerConsumerBase {
super.internalSetup();
super.producerBaseSetup();
- config.put(TaskConfig.TASK_CLASS_CONFIG,
org.apache.kafka.connect.file.FileStreamSourceTask.class);
- config.put(PulsarKafkaWorkerConfig.KEY_CONVERTER_CLASS_CONFIG,
org.apache.kafka.connect.storage.StringConverter.class);
- config.put(PulsarKafkaWorkerConfig.VALUE_CONVERTER_CLASS_CONFIG,
org.apache.kafka.connect.storage.StringConverter.class);
+ config.put(TaskConfig.TASK_CLASS_CONFIG,
"org.apache.kafka.connect.file.FileStreamSourceTask");
+ config.put(PulsarKafkaWorkerConfig.KEY_CONVERTER_CLASS_CONFIG,
"org.apache.kafka.connect.storage.StringConverter");
+ config.put(PulsarKafkaWorkerConfig.VALUE_CONVERTER_CLASS_CONFIG,
"org.apache.kafka.connect.storage.StringConverter");
this.offsetTopicName =
"persistent://my-property/my-ns/kafka-connect-source-offset";
config.put(PulsarKafkaWorkerConfig.PULSAR_SERVICE_URL_CONFIG,
brokerUrl.toString());
