support mysql binlog sync to pulsar by canal by tuteng · Pull Request #2998 · apache/pulsar

tuteng · 2018-11-16T11:03:42Z

Motivation

support alibaba canal
https://github.com/alibaba/canal/wiki

Modifications

Integrated canal client

Result

support binlog sync to pulsar
use python pulsar-client to consume

import pulsar

client = pulsar.Client('pulsar://localhost:6650')
consumer = client.subscribe('my-topic',
                            subscription_name='my-sub')

while True:
    msg = consumer.receive()
    print("Received message: '%s'" % msg.data())
    consumer.acknowledge(msg)

client.close()

output:
Received message: '[{"data":null,"database":"testdb","es":1542446501000,"id":44,"isDdl":true,"mysqlType":null,"old":null,"sql":"CREATE TABLE `users320` (   `id` int(11) NOT NULL AUTO_INCREMENT,   `name` varchar(50) DEFAULT NULL,   `extra` varchar(50) DEFAULT NULL,   PRIMARY KEY (`id`),   KEY `ix_users_name` (`name`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8","sqlType":null,"table":"users320","ts":1542446501114,"type":"CREATE"}]'

modify canal.client 1.1.1 for support flatMessage

…ql-binlog-canal

merlimat · 2018-11-16T17:14:56Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

+
+    protected void start() {
+        Assert.notNull(connector, "connector is null");
+        thread = new Thread(new Runnable() {


Maybe assign a name to thread so it's easier to debug with jstack

merlimat · 2018-11-16T17:15:43Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

@@ -0,0 +1,152 @@
+package org.apache.pulsar.io.canal;


This needs to have Apache headers. You can use mvn license:format to automatically add

sijie

overall looks good. it is a great contributions. have some review comments and also need some clarifications from you.

sijie · 2018-11-16T21:44:00Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

+
+        @Override
+        public void uncaughtException(Thread t, Throwable e) {
+            log.error("parse events has an error", e);


nit: it is good to add thread name in the logging message for troubleshooting purpose.

sijie · 2018-11-16T21:44:15Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

+
+    private CanalSourceConfig canalSourceConfig;
+
+    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {


protected final

sijie · 2018-11-16T21:45:36Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

+                    new InetSocketAddress(canalSourceConfig.getSingleHostname(), canalSourceConfig.getSinglePort()),
+                    canalSourceConfig.getDestination(), canalSourceConfig.getUsername(), canalSourceConfig.getPassword());
+        }
+        log.info("start canal connect");


can you improve the logging here?

in cluster mode, can you print out the zookeeper user and destination?

in single mode, can you print out the single host name, port and destination?

sijie · 2018-11-16T21:47:18Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

+    }
+
+    protected void start() {
+        Assert.notNull(connector, "connector is null");


nit: Use java builtin objects.requirenonnull or guava checkNotNull?

sijie · 2018-11-16T21:48:33Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

+            connector.disconnect();
+        }
+
+        MDC.remove("destination");


can you define destination as a constant string?

sijie · 2018-11-16T21:51:40Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

+    static private class CanalRecord implements Record<byte[]> {
+
+        private final byte[] record;
+        private final Long id;


is this id incremental? if you might consider using it as record sequence.

sijie · 2018-11-16T21:52:56Z

pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalSource.java

+                        }
+                    }
+
+                    connector.ack(batchId);


if you are acking here, it only supports at-most-once semantic. you might consider implementing ack() in CanalRecord

sijie · 2018-11-16T21:53:22Z

pulsar-io/pom.xml

    <module>elastic-search</module>
    <module>kafka-connect-adaptor</module>
    <module>debezium</module>
+      <module>canal</module>


nit: fix the indent here

add thread name add log info set to constant string destination Class CanalRecord add getRecordSequence

add Batch consume

Delete import Assert

tuteng · 2018-11-17T09:44:25Z

Please take the time to review my code again. It has been changed and tested locally.

sijie

great contribution! @tuteng

sijie · 2018-11-19T00:44:40Z

run integration tests

…om/AmateurEvents/pulsar into feature/support-mysql-binlog-canal

…ql-binlog-canal

sijie · 2018-11-22T01:36:38Z

run integration tests

2018-11-22\T\01:48:22.273 [INFO] Pulsar :: Distribution :: Offloader ................ SUCCESS [  3.790 s]
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Docker Images :: Pulsar Latest Version FAILURE [  9.247 s]
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Docker Images :: Grafana .......... SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Docker Images :: Pulsar Latest Version (Include All Components) SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Docker Images :: Pulsar Standalone  SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Tests ............................. SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Tests :: Docker Images ............ SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Tests :: Docker Images :: Latest Version Testing SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Tests :: Integration .............. SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Apache Pulsar :: Tests :: Pulsar Kafka Compat Client Tests SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Pulsar Storm adapter Tests ......................... SKIPPED
2018-11-22\T\01:48:22.274 [INFO] Spark Streaming Pulsar Receivers Tests ............. SKIPPED
2018-11-22\T\01:48:22.274 [INFO] ------------------------------------------------------------------------
2018-11-22\T\01:48:22.274 [INFO] BUILD FAILURE
2018-11-22\T\01:48:22.275 [INFO] ------------------------------------------------------------------------
2018-11-22\T\01:48:22.275 [INFO] Total time: 11:02 min
2018-11-22\T\01:48:22.275 [INFO] Finished at: 2018-11-22T01:48:22Z
2018-11-22\T\01:48:24.095 [INFO] Final Memory: 293M/3757M
2018-11-22\T\01:48:24.095 [INFO] ------------------------------------------------------------------------
2018-11-22\T\01:48:24.102 [ERROR] Failed to execute goal org.codehaus.mojo:exec-maven-plugin:1.6.0:exec (build-pulsar-clients) on project pulsar-docker-image: Command execution failed.: Process exited with an error: 1 (Exit value: 1) -> [Help 1]
2018-11-22\T\01:48:24.103 [ERROR] 
2018-11-22\T\01:48:24.103 [ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.
2018-11-22\T\01:48:24.103 [ERROR] Re-run Maven using the -X switch to enable full debug logging.
2018-11-22\T\01:48:24.103 [ERROR]

jiazhai · 2018-11-22T09:50:58Z

run integration tests

org.apache.pulsar.tests.integration.offload.TestS3Offload.tiered-storage-test-suite

jiazhai · 2018-11-22T12:09:13Z

run integration tests

org.apache.pulsar.tests.integration.offload.TestS3Offload.tiered-storage-test-suite

jiazhai

lgtm. Thanks for the contribute.

jiazhai · 2018-11-22T13:17:46Z

run integration tests
org.apache.pulsar.tests.integration.offload.TestS3Offload.tiered-storage-test-suite

jiazhai · 2018-11-23T06:37:53Z

run integration tests

jiazhai · 2018-11-25T04:32:47Z

run integration tests

jiazhai · 2018-11-25T08:36:23Z

run integration tests

sijie · 2018-11-25T19:40:58Z

@tuteng thank you so much for your contribution.

tuteng and others added 6 commits November 11, 2018 20:50

support canal for mysql binlog to pulsar

6b1aa50

modify message to []byte for consumer

e281f34

modify canal.client 1.1.1 for support flatMessage

Merge remote-tracking branch 'apache/master' into feature/support-mys…

3679bb7

…ql-binlog-canal

modify sourceClass

f34dc39

Delete duplicate dependency

726571f

add log

c1a8733

merlimat reviewed Nov 16, 2018

View reviewed changes

sijie assigned tuteng Nov 16, 2018

sijie requested a review from jiazhai November 16, 2018 21:42

sijie added type/feature The PR added a new feature or issue requested a new feature area/connector labels Nov 16, 2018

sijie added this to the 2.3.0 milestone Nov 16, 2018

sijie reviewed Nov 16, 2018

View reviewed changes

tuteng added 9 commits November 17, 2018 11:13

add license

8264b15

add thread name add log info set to constant string destination Class CanalRecord add getRecordSequence

add log destination

3022916

override ack

3e6914e

format pom file

f22f061

add consume to pulsar

34b77a4

Delete duplicate ack

75b70bb

add Batch consume

Log batchId in ack

b4d6625

Log thread name in uncaughtException

ac21cbd

Delete import Assert

format log

8b0f4eb

sijie approved these changes Nov 19, 2018

View reviewed changes

sijie added 2 commits November 21, 2018 11:25

Merge branch 'feature/support-mysql-binlog-canal' of https://github.c…

886aa4b

…om/AmateurEvents/pulsar into feature/support-mysql-binlog-canal

Merge remote-tracking branch 'apache/master' into feature/support-mys…

0a8e307

…ql-binlog-canal

jiazhai approved these changes Nov 22, 2018

View reviewed changes

jiazhai mentioned this pull request Nov 22, 2018

Intermittent test failure in TestS3Offload #2942

Closed

sijie merged commit c2c9f33 into apache:master Nov 25, 2018


		private CanalSourceConfig canalSourceConfig;

		protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {

Conversation

tuteng commented Nov 16, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Result

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sijie left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tuteng commented Nov 17, 2018

Uh oh!

sijie left a comment

Choose a reason for hiding this comment

Uh oh!

sijie commented Nov 19, 2018

Uh oh!

sijie commented Nov 22, 2018 • edited by jiazhai Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jiazhai commented Nov 22, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jiazhai commented Nov 22, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jiazhai left a comment

Choose a reason for hiding this comment

Uh oh!

jiazhai commented Nov 22, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jiazhai commented Nov 23, 2018

Uh oh!

jiazhai commented Nov 25, 2018

Uh oh!

jiazhai commented Nov 25, 2018

Uh oh!

sijie commented Nov 25, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

tuteng commented Nov 16, 2018 •

edited

Loading

sijie commented Nov 22, 2018 •

edited by jiazhai

Loading

jiazhai commented Nov 22, 2018 •

edited

Loading

jiazhai commented Nov 22, 2018 •

edited

Loading

jiazhai commented Nov 22, 2018 •

edited

Loading