You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2022/05/07 11:44:04 UTC
[GitHub] [incubator-inlong] Greedyu opened a new pull request, #4113: [INLONG-4112][Agent]Feature binlog source start special
Greedyu opened a new pull request, #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113
[INLONG-4112][Agent] mysql binlog supports collection from a specified start
#4112
### Motivation
1. mysql binlog supports collection from a specified start
2. Fix BinlogReader init() bug
### Modifications
Refer to flink to specify the location reading method。The core principle is to modify the OffsetBackingStore object when initializing the OffsetStorageWriter class in the configure() of the FlinkOffsetBackingStore class. When specifying the location, the "offset.storage" configuration here directly uses FlinkOffsetBackingStore.class.getCanonicalName(), and specifies FlinkOffsetBackingStore.OFFSET_STATE_VALUE, "snapshot.mode" and other parameters according to the flink configuration
### Verifying this change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] EMsnap commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
EMsnap commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r875402890
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestBinlogReader.java:
##########
@@ -18,43 +18,74 @@
package org.apache.inlong.agent.plugin.sources;
import com.google.gson.Gson;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.constant.SnapshotModeConstants;
+import org.apache.inlong.agent.plugin.sources.reader.BinlogReader;
import org.apache.inlong.agent.pojo.DebeziumFormat;
import org.junit.Assert;
import org.junit.Test;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
+
public class TestBinlogReader {
private static Gson gson = new Gson();
@Test
public void testDebeziumFormat() {
String debeziumJson = "{\n"
- + " \"before\": null,\n"
- + " \"after\": {\n"
- + " \"id\": 1004,\n"
- + " \"first_name\": \"Anne\",\n"
- + " \"last_name\": \"Kretchmar\",\n"
- + " \"email\": \"annek@noanswer.org\"\n"
- + " },\n"
- + " \"source\": {\n"
- + " \"version\": \"1.8.1.Final\",\n"
- + " \"name\": \"dbserver1\",\n"
- + " \"server_id\": 0,\n"
- + " \"ts_sec\": 0,\n"
- + " \"gtid\": null,\n"
- + " \"file\": \"mysql-bin.000003\",\n"
- + " \"pos\": 154,\n"
- + " \"row\": 0,\n"
- + " \"snapshot\": true,\n"
- + " \"thread\": null,\n"
- + " \"db\": \"inventory\",\n"
- + " \"table\": \"customers\"\n"
- + " },\n"
- + " \"op\": \"r\",\n"
- + " \"ts_ms\": 1486500577691\n"
- + " }";
+ + " \"before\": null,\n"
+ + " \"after\": {\n"
+ + " \"id\": 1004,\n"
+ + " \"first_name\": \"Anne\",\n"
+ + " \"last_name\": \"Kretchmar\",\n"
+ + " \"email\": \"annek@noanswer.org\"\n"
+ + " },\n"
+ + " \"source\": {\n"
+ + " \"version\": \"1.8.1.Final\",\n"
+ + " \"name\": \"dbserver1\",\n"
+ + " \"server_id\": 0,\n"
+ + " \"ts_sec\": 0,\n"
+ + " \"gtid\": null,\n"
+ + " \"file\": \"mysql-bin.000003\",\n"
+ + " \"pos\": 154,\n"
+ + " \"row\": 0,\n"
+ + " \"snapshot\": true,\n"
+ + " \"thread\": null,\n"
+ + " \"db\": \"inventory\",\n"
+ + " \"table\": \"customers\"\n"
+ + " },\n"
+ + " \"op\": \"r\",\n"
+ + " \"ts_ms\": 1486500577691\n"
+ + " }";
DebeziumFormat debeziumFormat = gson
- .fromJson(debeziumJson, DebeziumFormat.class);
+ .fromJson(debeziumJson, DebeziumFormat.class);
Assert.assertEquals("customers", debeziumFormat.getSource().getTable());
}
+
+ // @Test
+ public void binlogStartSpacialTest() throws Exception {
+ JobProfile jobProfile = new JobProfile();
+ jobProfile.set(BinlogReader.JOB_DATABASE_USER, "root");
+ jobProfile.set(BinlogReader.JOB_DATABASE_PASSWORD, "123456");
+ jobProfile.set(BinlogReader.JOB_DATABASE_HOSTNAME, "");
+ jobProfile.set(BinlogReader.JOB_DATABASE_PORT, "3307");
+ jobProfile.set(BinlogReader.JOB_DATABASE_WHITELIST, "etl");
+ jobProfile.set(BinlogReader.JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE, "mysql-bin.000001");
+ jobProfile.set(BinlogReader.JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_POS, "5641");
+ // jobProfile.set(BinlogReader.JOB_DATABASE_SNAPSHOT_MODE, SnapshotModeConstants.INITIAL);
+ // jobProfile.set(BinlogReader.JOB_DATABASE_SNAPSHOT_MODE, SnapshotModeConstants.EARLIEST_OFFSET);
Review Comment:
do we need this line of code ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r874724236
##########
inlong-agent/agent-plugins/pom.xml:
##########
@@ -69,6 +69,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
Review Comment:
Because the debezium-embedded package calls kafka client 2.4.1, but the Utils.isBlank method is connect-runtime 3.0.0. There is a problem with this online logic, it is a bug
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java:
##########
@@ -41,38 +33,54 @@
import org.apache.inlong.agent.plugin.Message;
import org.apache.inlong.agent.plugin.Reader;
import org.apache.inlong.agent.plugin.sources.snapshot.BinlogSnapshotBase;
+import org.apache.inlong.agent.plugin.utils.InLongDatabaseHistory;
+import org.apache.inlong.agent.plugin.utils.InLongOffsetBackingStore;
import org.apache.inlong.agent.pojo.DebeziumFormat;
+import org.apache.inlong.agent.pojo.DebeziumOffset;
import org.apache.inlong.agent.utils.AgentUtils;
+import org.apache.inlong.agent.utils.DebeziumOffsetSerializer;
import org.apache.inlong.common.reporpter.ConfigLogTypeEnum;
import org.apache.inlong.common.reporpter.StreamConfigLogMetric;
import org.apache.kafka.connect.storage.FileOffsetBackingStore;
import org.slf4j.Logger;
-
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_MAP_CAPACITY;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_DATA;
+
public class BinlogReader implements Reader {
private static final Logger LOGGER = LoggerFactory.getLogger(BinlogReader.class);
- private static final String COMPONENT_NAME = "BinlogReader";
- private static final String JOB_DATABASE_USER = "job.binlogJob.user";
- private static final String JOB_DATABASE_PASSWORD = "job.binlogJob.password";
- private static final String JOB_DATABASE_HOSTNAME = "job.binlogJob.hostname";
- private static final String JOB_TABLE_WHITELIST = "job.binlogJob.tableWhiteList";
- private static final String JOB_DATABASE_WHITELIST = "job.binlogJob.databaseWhiteList";
+ public static final String COMPONENT_NAME = "BinlogReader";
+ public static final String JOB_DATABASE_USER = "job.binlogJob.user";
+ public static final String JOB_DATABASE_PASSWORD = "job.binlogJob.password";
+ public static final String JOB_DATABASE_HOSTNAME = "job.binlogJob.hostname";
+ public static final String JOB_TABLE_WHITELIST = "job.binlogJob.tableWhiteList";
+ public static final String JOB_DATABASE_WHITELIST = "job.binlogJob.databaseWhiteList";
- private static final String JOB_DATABASE_OFFSETS = "job.binlogJob.offsets";
- private static final String JOB_DATABASE_OFFSET_FILENAME = "job.binlogJob.offset.filename";
+ public static final String JOB_DATABASE_OFFSETS = "job.binlogJob.offsets";
+ public static final String JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE = "job.binlogJob.offset.specificOffsetFile";
+ public static final String JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_POS = "job.binlogJob.offset.specificOffsetPos";
- private static final String JOB_DATABASE_SERVER_TIME_ZONE = "job.binlogJob.serverTimezone";
- private static final String JOB_DATABASE_STORE_OFFSET_INTERVAL_MS = "job.binlogJob.offset.intervalMs";
+ public static final String JOB_DATABASE_SERVER_TIME_ZONE = "job.binlogJob.serverTimezone";
Review Comment:
unit test use
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r869052840
##########
inlong-agent/agent-plugins/pom.xml:
##########
@@ -79,6 +84,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>flink-connector-mysql-cdc</artifactId>
+ <version>${flink.connector.mysql.cdc.version}</version>
Review Comment:
resolve
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] EMsnap commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
EMsnap commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r867601844
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java:
##########
@@ -253,6 +271,28 @@ private Properties getEngineProps() {
return props;
}
+ private String serializeOffset() {
+ Map<String, Object> sourceOffset = new HashMap<>();
+ Preconditions.checkNotNull(JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE,
+ JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE + "shouldn't be null");
+ sourceOffset.put("file", specificOffsetFile);
+ Preconditions.checkNotNull(JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_POS,
+ JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_POS + " shouldn't be null");
+ sourceOffset.put("pos", specificOffsetPos);
+ DebeziumOffset specificOffset = new DebeziumOffset();
+ specificOffset.setSourceOffset(sourceOffset);
+ Map<String, String> sourcePartition = new HashMap<>();
+ sourcePartition.put("server", instanceId);
+ specificOffset.setSourcePartition(sourcePartition);
+ byte[] serializedOffset = new byte[0];
+ try {
+ serializedOffset = DebeziumOffsetSerializer.INSTANCE.serialize(specificOffset);
Review Comment:
Here we can actually implement a DebeziumOffsetSerializer by ourselves
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java:
##########
@@ -228,18 +240,24 @@ private Properties getEngineProps() {
props.setProperty("table.whitelist", tableWhiteList);
props.setProperty("database.whitelist", databaseWhiteList);
- props.setProperty("offset.storage", FileOffsetBackingStore.class.getCanonicalName());
- props.setProperty("offset.storage.file.filename", offsetStoreFileName);
props.setProperty("offset.flush.interval.ms", offsetFlushIntervalMs);
- props.setProperty("database.history", FileDatabaseHistory.class.getCanonicalName());
- props.setProperty("database.history.file.filename", databaseStoreHistoryName);
props.setProperty("database.snapshot.mode", snapshotMode);
props.setProperty("database.history.store.only.monitored.tables.ddl", historyMonitorDdl);
props.setProperty("database.allowPublicKeyRetrieval", "true");
props.setProperty("key.converter.schemas.enable", "false");
props.setProperty("value.converter.schemas.enable", "false");
props.setProperty("include.schema.changes", includeSchemaChanges);
props.setProperty("snapshot.mode", snapshotMode);
+ if ("schema_only_recovery".equals(snapshotMode)) {
+ props.setProperty("offset.storage", FlinkOffsetBackingStore.class.getCanonicalName());
+ props.setProperty(FlinkOffsetBackingStore.OFFSET_STATE_VALUE, serializeOffset());
+ props.setProperty("database.history", FlinkDatabaseHistory.class.getCanonicalName());
+ } else {
+ props.setProperty("offset.storage", FileOffsetBackingStore.class.getCanonicalName());
Review Comment:
no to to include FileDatabaseHistory and FileOffsetBackingStore here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] EMsnap commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
EMsnap commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r871198962
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/DatabaseHistoryUtil.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.inlong.agent.plugin.utils;
+
+import org.apache.inlong.agent.plugin.message.SchemaRecord;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @description
+ * @date: 2022/5/10
Review Comment:
please remove this or update with some comments
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/SchemaRecord.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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.inlong.agent.plugin.message;
+
+import io.debezium.document.Document;
+import io.debezium.document.DocumentWriter;
+import io.debezium.relational.history.HistoryRecord;
+
+import java.io.IOException;
+
+/**
+ * @description
Review Comment:
please remove this or update with some comments
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java:
##########
@@ -41,38 +33,54 @@
import org.apache.inlong.agent.plugin.Message;
import org.apache.inlong.agent.plugin.Reader;
import org.apache.inlong.agent.plugin.sources.snapshot.BinlogSnapshotBase;
+import org.apache.inlong.agent.plugin.utils.InLongDatabaseHistory;
+import org.apache.inlong.agent.plugin.utils.InLongOffsetBackingStore;
import org.apache.inlong.agent.pojo.DebeziumFormat;
+import org.apache.inlong.agent.pojo.DebeziumOffset;
import org.apache.inlong.agent.utils.AgentUtils;
+import org.apache.inlong.agent.utils.DebeziumOffsetSerializer;
import org.apache.inlong.common.reporpter.ConfigLogTypeEnum;
import org.apache.inlong.common.reporpter.StreamConfigLogMetric;
import org.apache.kafka.connect.storage.FileOffsetBackingStore;
import org.slf4j.Logger;
-
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.apache.inlong.agent.constant.CommonConstants.DEFAULT_MAP_CAPACITY;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_KEY_DATA;
+
public class BinlogReader implements Reader {
private static final Logger LOGGER = LoggerFactory.getLogger(BinlogReader.class);
- private static final String COMPONENT_NAME = "BinlogReader";
- private static final String JOB_DATABASE_USER = "job.binlogJob.user";
- private static final String JOB_DATABASE_PASSWORD = "job.binlogJob.password";
- private static final String JOB_DATABASE_HOSTNAME = "job.binlogJob.hostname";
- private static final String JOB_TABLE_WHITELIST = "job.binlogJob.tableWhiteList";
- private static final String JOB_DATABASE_WHITELIST = "job.binlogJob.databaseWhiteList";
+ public static final String COMPONENT_NAME = "BinlogReader";
+ public static final String JOB_DATABASE_USER = "job.binlogJob.user";
+ public static final String JOB_DATABASE_PASSWORD = "job.binlogJob.password";
+ public static final String JOB_DATABASE_HOSTNAME = "job.binlogJob.hostname";
+ public static final String JOB_TABLE_WHITELIST = "job.binlogJob.tableWhiteList";
+ public static final String JOB_DATABASE_WHITELIST = "job.binlogJob.databaseWhiteList";
- private static final String JOB_DATABASE_OFFSETS = "job.binlogJob.offsets";
- private static final String JOB_DATABASE_OFFSET_FILENAME = "job.binlogJob.offset.filename";
+ public static final String JOB_DATABASE_OFFSETS = "job.binlogJob.offsets";
+ public static final String JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE = "job.binlogJob.offset.specificOffsetFile";
+ public static final String JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_POS = "job.binlogJob.offset.specificOffsetPos";
- private static final String JOB_DATABASE_SERVER_TIME_ZONE = "job.binlogJob.serverTimezone";
- private static final String JOB_DATABASE_STORE_OFFSET_INTERVAL_MS = "job.binlogJob.offset.intervalMs";
+ public static final String JOB_DATABASE_SERVER_TIME_ZONE = "job.binlogJob.serverTimezone";
Review Comment:
why public this?
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/InLongDatabaseHistory.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.inlong.agent.plugin.utils;
+
+import io.debezium.config.Configuration;
+import io.debezium.relational.history.AbstractDatabaseHistory;
+import io.debezium.relational.history.DatabaseHistoryException;
+import io.debezium.relational.history.DatabaseHistoryListener;
+import io.debezium.relational.history.HistoryRecord;
+import io.debezium.relational.history.HistoryRecordComparator;
+import org.apache.inlong.agent.plugin.message.SchemaRecord;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.function.Consumer;
+
+/**
+ * @description
Review Comment:
ditto
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/InLongDatabaseHistory.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.inlong.agent.plugin.utils;
+
+import io.debezium.config.Configuration;
+import io.debezium.relational.history.AbstractDatabaseHistory;
+import io.debezium.relational.history.DatabaseHistoryException;
+import io.debezium.relational.history.DatabaseHistoryListener;
+import io.debezium.relational.history.HistoryRecord;
+import io.debezium.relational.history.HistoryRecordComparator;
+import org.apache.inlong.agent.plugin.message.SchemaRecord;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.function.Consumer;
+
+/**
+ * @description
Review Comment:
remove the date also please
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r874725068
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java:
##########
@@ -152,12 +162,14 @@ public void init(JobProfile jobConf) {
finished = false;
offset = jobConf.get(JOB_DATABASE_OFFSETS, "");
+ specificOffsetFile = jobConf.get(JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE, "");
Review Comment:
next commit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r867864628
##########
inlong-agent/agent-plugins/pom.xml:
##########
@@ -29,6 +29,13 @@
<modelVersion>4.0.0</modelVersion>
<name>Apache InLong - Agent Plugins</name>
+ <properties>
+ <kafka.clients.version>3.0.0</kafka.clients.version>
+ <mysql-connector-java.version>8.0.27</mysql-connector-java.version>
+ <flink-shaded-jackson.version>2.12.1-13.0</flink-shaded-jackson.version>
+ <agent-flink.connector.mysql.cdc.version>2.2.0</agent-flink.connector.mysql.cdc.version>
+ </properties>
Review Comment:
resolve
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] dockerzhang commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
dockerzhang commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r868954004
##########
inlong-agent/agent-plugins/pom.xml:
##########
@@ -79,6 +84,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>flink-connector-mysql-cdc</artifactId>
+ <version>${flink.connector.mysql.cdc.version}</version>
Review Comment:
it's better to avoid using the `flink-connector-mysql-cdc` dependency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] dockerzhang merged pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
dockerzhang merged PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] EMsnap commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
EMsnap commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r871235156
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java:
##########
@@ -152,12 +162,14 @@ public void init(JobProfile jobConf) {
finished = false;
offset = jobConf.get(JOB_DATABASE_OFFSETS, "");
+ specificOffsetFile = jobConf.get(JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE, "");
Review Comment:
please add the parameter in inlong manager and inlong manager client
##########
inlong-agent/agent-plugins/pom.xml:
##########
@@ -69,6 +69,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
Review Comment:
why kafka clients here ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] EMsnap commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
EMsnap commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r875423094
##########
pom.xml:
##########
@@ -229,6 +229,8 @@
<tencentcloud.cls.version>1.0.5</tencentcloud.cls.version>
<esri-geometry-api.version>2.0.0</esri-geometry-api.version>
<HikariCP.version>4.0.3</HikariCP.version>
+
+ <kafka.clients.version>3.0.0</kafka.clients.version>
Review Comment:
please add notice and dependency
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r867866214
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java:
##########
@@ -253,6 +271,28 @@ private Properties getEngineProps() {
return props;
}
+ private String serializeOffset() {
+ Map<String, Object> sourceOffset = new HashMap<>();
+ Preconditions.checkNotNull(JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE,
+ JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE + "shouldn't be null");
+ sourceOffset.put("file", specificOffsetFile);
+ Preconditions.checkNotNull(JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_POS,
+ JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_POS + " shouldn't be null");
+ sourceOffset.put("pos", specificOffsetPos);
+ DebeziumOffset specificOffset = new DebeziumOffset();
+ specificOffset.setSourceOffset(sourceOffset);
+ Map<String, String> sourcePartition = new HashMap<>();
+ sourcePartition.put("server", instanceId);
+ specificOffset.setSourcePartition(sourcePartition);
+ byte[] serializedOffset = new byte[0];
+ try {
+ serializedOffset = DebeziumOffsetSerializer.INSTANCE.serialize(specificOffset);
Review Comment:
Create new class org.apache.inlong.agent.utils.DebeziumOffsetSerializer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r867867078
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java:
##########
@@ -228,18 +240,24 @@ private Properties getEngineProps() {
props.setProperty("table.whitelist", tableWhiteList);
props.setProperty("database.whitelist", databaseWhiteList);
- props.setProperty("offset.storage", FileOffsetBackingStore.class.getCanonicalName());
- props.setProperty("offset.storage.file.filename", offsetStoreFileName);
props.setProperty("offset.flush.interval.ms", offsetFlushIntervalMs);
- props.setProperty("database.history", FileDatabaseHistory.class.getCanonicalName());
- props.setProperty("database.history.file.filename", databaseStoreHistoryName);
props.setProperty("database.snapshot.mode", snapshotMode);
props.setProperty("database.history.store.only.monitored.tables.ddl", historyMonitorDdl);
props.setProperty("database.allowPublicKeyRetrieval", "true");
props.setProperty("key.converter.schemas.enable", "false");
props.setProperty("value.converter.schemas.enable", "false");
props.setProperty("include.schema.changes", includeSchemaChanges);
props.setProperty("snapshot.mode", snapshotMode);
+ if ("schema_only_recovery".equals(snapshotMode)) {
+ props.setProperty("offset.storage", FlinkOffsetBackingStore.class.getCanonicalName());
+ props.setProperty(FlinkOffsetBackingStore.OFFSET_STATE_VALUE, serializeOffset());
+ props.setProperty("database.history", FlinkDatabaseHistory.class.getCanonicalName());
+ } else {
+ props.setProperty("offset.storage", FileOffsetBackingStore.class.getCanonicalName());
Review Comment:
Create new class org.apache.inlong.agent.plugin.utils.InLongDatabaseHistory, org.apache.inlong.agent.plugin.utils.InLongOffsetBackingStore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r867864911
##########
inlong-agent/agent-plugins/pom.xml:
##########
@@ -69,6 +76,17 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.clients.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>${mysql-connector-java.version}</version>
+ <scope>runtime</scope>
Review Comment:
resolve
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestBinlogReader.java:
##########
@@ -18,43 +18,70 @@
package org.apache.inlong.agent.plugin.sources;
import com.google.gson.Gson;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.plugin.sources.reader.BinlogReader;
import org.apache.inlong.agent.pojo.DebeziumFormat;
import org.junit.Assert;
import org.junit.Test;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
+
public class TestBinlogReader {
private static Gson gson = new Gson();
@Test
public void testDebeziumFormat() {
String debeziumJson = "{\n"
- + " \"before\": null,\n"
- + " \"after\": {\n"
- + " \"id\": 1004,\n"
- + " \"first_name\": \"Anne\",\n"
- + " \"last_name\": \"Kretchmar\",\n"
- + " \"email\": \"annek@noanswer.org\"\n"
- + " },\n"
- + " \"source\": {\n"
- + " \"version\": \"1.8.1.Final\",\n"
- + " \"name\": \"dbserver1\",\n"
- + " \"server_id\": 0,\n"
- + " \"ts_sec\": 0,\n"
- + " \"gtid\": null,\n"
- + " \"file\": \"mysql-bin.000003\",\n"
- + " \"pos\": 154,\n"
- + " \"row\": 0,\n"
- + " \"snapshot\": true,\n"
- + " \"thread\": null,\n"
- + " \"db\": \"inventory\",\n"
- + " \"table\": \"customers\"\n"
- + " },\n"
- + " \"op\": \"r\",\n"
- + " \"ts_ms\": 1486500577691\n"
- + " }";
+ + " \"before\": null,\n"
+ + " \"after\": {\n"
+ + " \"id\": 1004,\n"
+ + " \"first_name\": \"Anne\",\n"
+ + " \"last_name\": \"Kretchmar\",\n"
+ + " \"email\": \"annek@noanswer.org\"\n"
+ + " },\n"
+ + " \"source\": {\n"
+ + " \"version\": \"1.8.1.Final\",\n"
+ + " \"name\": \"dbserver1\",\n"
+ + " \"server_id\": 0,\n"
+ + " \"ts_sec\": 0,\n"
+ + " \"gtid\": null,\n"
+ + " \"file\": \"mysql-bin.000003\",\n"
+ + " \"pos\": 154,\n"
+ + " \"row\": 0,\n"
+ + " \"snapshot\": true,\n"
+ + " \"thread\": null,\n"
+ + " \"db\": \"inventory\",\n"
+ + " \"table\": \"customers\"\n"
+ + " },\n"
+ + " \"op\": \"r\",\n"
+ + " \"ts_ms\": 1486500577691\n"
+ + " }";
DebeziumFormat debeziumFormat = gson
- .fromJson(debeziumJson, DebeziumFormat.class);
+ .fromJson(debeziumJson, DebeziumFormat.class);
Assert.assertEquals("customers", debeziumFormat.getSource().getTable());
}
+
+ // @Test
+ public void binlogStartSpacialTest() throws Exception {
+ JobProfile jobProfile = new JobProfile();
+ jobProfile.set(BinlogReader.JOB_DATABASE_USER, "root");
+ jobProfile.set(BinlogReader.JOB_DATABASE_PASSWORD, "123456");
+ jobProfile.set(BinlogReader.JOB_DATABASE_HOSTNAME, "9.135.12.10");
Review Comment:
resolve
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] dockerzhang commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
dockerzhang commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r867598547
##########
inlong-agent/agent-plugins/pom.xml:
##########
@@ -69,6 +76,17 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.clients.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>${mysql-connector-java.version}</version>
+ <scope>runtime</scope>
Review Comment:
due to the license reason, we should not add the MySQL dependency.
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestBinlogReader.java:
##########
@@ -18,43 +18,70 @@
package org.apache.inlong.agent.plugin.sources;
import com.google.gson.Gson;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.plugin.sources.reader.BinlogReader;
import org.apache.inlong.agent.pojo.DebeziumFormat;
import org.junit.Assert;
import org.junit.Test;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
+
public class TestBinlogReader {
private static Gson gson = new Gson();
@Test
public void testDebeziumFormat() {
String debeziumJson = "{\n"
- + " \"before\": null,\n"
- + " \"after\": {\n"
- + " \"id\": 1004,\n"
- + " \"first_name\": \"Anne\",\n"
- + " \"last_name\": \"Kretchmar\",\n"
- + " \"email\": \"annek@noanswer.org\"\n"
- + " },\n"
- + " \"source\": {\n"
- + " \"version\": \"1.8.1.Final\",\n"
- + " \"name\": \"dbserver1\",\n"
- + " \"server_id\": 0,\n"
- + " \"ts_sec\": 0,\n"
- + " \"gtid\": null,\n"
- + " \"file\": \"mysql-bin.000003\",\n"
- + " \"pos\": 154,\n"
- + " \"row\": 0,\n"
- + " \"snapshot\": true,\n"
- + " \"thread\": null,\n"
- + " \"db\": \"inventory\",\n"
- + " \"table\": \"customers\"\n"
- + " },\n"
- + " \"op\": \"r\",\n"
- + " \"ts_ms\": 1486500577691\n"
- + " }";
+ + " \"before\": null,\n"
+ + " \"after\": {\n"
+ + " \"id\": 1004,\n"
+ + " \"first_name\": \"Anne\",\n"
+ + " \"last_name\": \"Kretchmar\",\n"
+ + " \"email\": \"annek@noanswer.org\"\n"
+ + " },\n"
+ + " \"source\": {\n"
+ + " \"version\": \"1.8.1.Final\",\n"
+ + " \"name\": \"dbserver1\",\n"
+ + " \"server_id\": 0,\n"
+ + " \"ts_sec\": 0,\n"
+ + " \"gtid\": null,\n"
+ + " \"file\": \"mysql-bin.000003\",\n"
+ + " \"pos\": 154,\n"
+ + " \"row\": 0,\n"
+ + " \"snapshot\": true,\n"
+ + " \"thread\": null,\n"
+ + " \"db\": \"inventory\",\n"
+ + " \"table\": \"customers\"\n"
+ + " },\n"
+ + " \"op\": \"r\",\n"
+ + " \"ts_ms\": 1486500577691\n"
+ + " }";
DebeziumFormat debeziumFormat = gson
- .fromJson(debeziumJson, DebeziumFormat.class);
+ .fromJson(debeziumJson, DebeziumFormat.class);
Assert.assertEquals("customers", debeziumFormat.getSource().getTable());
}
+
+ // @Test
+ public void binlogStartSpacialTest() throws Exception {
+ JobProfile jobProfile = new JobProfile();
+ jobProfile.set(BinlogReader.JOB_DATABASE_USER, "root");
+ jobProfile.set(BinlogReader.JOB_DATABASE_PASSWORD, "123456");
+ jobProfile.set(BinlogReader.JOB_DATABASE_HOSTNAME, "9.135.12.10");
Review Comment:
remove the IP, it's better to mock the UT.
##########
inlong-agent/agent-plugins/pom.xml:
##########
@@ -29,6 +29,13 @@
<modelVersion>4.0.0</modelVersion>
<name>Apache InLong - Agent Plugins</name>
+ <properties>
+ <kafka.clients.version>3.0.0</kafka.clients.version>
+ <mysql-connector-java.version>8.0.27</mysql-connector-java.version>
+ <flink-shaded-jackson.version>2.12.1-13.0</flink-shaded-jackson.version>
+ <agent-flink.connector.mysql.cdc.version>2.2.0</agent-flink.connector.mysql.cdc.version>
+ </properties>
Review Comment:
please move the version definition to the root pom.xml
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r875443073
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestBinlogReader.java:
##########
@@ -18,43 +18,74 @@
package org.apache.inlong.agent.plugin.sources;
import com.google.gson.Gson;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.constant.SnapshotModeConstants;
+import org.apache.inlong.agent.plugin.sources.reader.BinlogReader;
import org.apache.inlong.agent.pojo.DebeziumFormat;
import org.junit.Assert;
import org.junit.Test;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_GROUP_ID;
+import static org.apache.inlong.agent.constant.CommonConstants.PROXY_INLONG_STREAM_ID;
+
public class TestBinlogReader {
private static Gson gson = new Gson();
@Test
public void testDebeziumFormat() {
String debeziumJson = "{\n"
- + " \"before\": null,\n"
- + " \"after\": {\n"
- + " \"id\": 1004,\n"
- + " \"first_name\": \"Anne\",\n"
- + " \"last_name\": \"Kretchmar\",\n"
- + " \"email\": \"annek@noanswer.org\"\n"
- + " },\n"
- + " \"source\": {\n"
- + " \"version\": \"1.8.1.Final\",\n"
- + " \"name\": \"dbserver1\",\n"
- + " \"server_id\": 0,\n"
- + " \"ts_sec\": 0,\n"
- + " \"gtid\": null,\n"
- + " \"file\": \"mysql-bin.000003\",\n"
- + " \"pos\": 154,\n"
- + " \"row\": 0,\n"
- + " \"snapshot\": true,\n"
- + " \"thread\": null,\n"
- + " \"db\": \"inventory\",\n"
- + " \"table\": \"customers\"\n"
- + " },\n"
- + " \"op\": \"r\",\n"
- + " \"ts_ms\": 1486500577691\n"
- + " }";
+ + " \"before\": null,\n"
+ + " \"after\": {\n"
+ + " \"id\": 1004,\n"
+ + " \"first_name\": \"Anne\",\n"
+ + " \"last_name\": \"Kretchmar\",\n"
+ + " \"email\": \"annek@noanswer.org\"\n"
+ + " },\n"
+ + " \"source\": {\n"
+ + " \"version\": \"1.8.1.Final\",\n"
+ + " \"name\": \"dbserver1\",\n"
+ + " \"server_id\": 0,\n"
+ + " \"ts_sec\": 0,\n"
+ + " \"gtid\": null,\n"
+ + " \"file\": \"mysql-bin.000003\",\n"
+ + " \"pos\": 154,\n"
+ + " \"row\": 0,\n"
+ + " \"snapshot\": true,\n"
+ + " \"thread\": null,\n"
+ + " \"db\": \"inventory\",\n"
+ + " \"table\": \"customers\"\n"
+ + " },\n"
+ + " \"op\": \"r\",\n"
+ + " \"ts_ms\": 1486500577691\n"
+ + " }";
DebeziumFormat debeziumFormat = gson
- .fromJson(debeziumJson, DebeziumFormat.class);
+ .fromJson(debeziumJson, DebeziumFormat.class);
Assert.assertEquals("customers", debeziumFormat.getSource().getTable());
}
+
+ // @Test
+ public void binlogStartSpacialTest() throws Exception {
+ JobProfile jobProfile = new JobProfile();
+ jobProfile.set(BinlogReader.JOB_DATABASE_USER, "root");
+ jobProfile.set(BinlogReader.JOB_DATABASE_PASSWORD, "123456");
+ jobProfile.set(BinlogReader.JOB_DATABASE_HOSTNAME, "");
+ jobProfile.set(BinlogReader.JOB_DATABASE_PORT, "3307");
+ jobProfile.set(BinlogReader.JOB_DATABASE_WHITELIST, "etl");
+ jobProfile.set(BinlogReader.JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_FILE, "mysql-bin.000001");
+ jobProfile.set(BinlogReader.JOB_DATABASE_OFFSET_SPECIFIC_OFFSET_POS, "5641");
+ // jobProfile.set(BinlogReader.JOB_DATABASE_SNAPSHOT_MODE, SnapshotModeConstants.INITIAL);
+ // jobProfile.set(BinlogReader.JOB_DATABASE_SNAPSHOT_MODE, SnapshotModeConstants.EARLIEST_OFFSET);
Review Comment:
resolve
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-inlong] Greedyu commented on a diff in pull request #4113: [INLONG-4112][Agent] Support collect data from the specified position for MySQL binlog
Posted by GitBox <gi...@apache.org>.
Greedyu commented on code in PR #4113:
URL: https://github.com/apache/incubator-inlong/pull/4113#discussion_r874721257
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/message/SchemaRecord.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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.inlong.agent.plugin.message;
+
+import io.debezium.document.Document;
+import io.debezium.document.DocumentWriter;
+import io.debezium.relational.history.HistoryRecord;
+
+import java.io.IOException;
+
+/**
+ * @description
Review Comment:
resolve
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/DatabaseHistoryUtil.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.inlong.agent.plugin.utils;
+
+import org.apache.inlong.agent.plugin.message.SchemaRecord;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @description
+ * @date: 2022/5/10
Review Comment:
resolve
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/utils/InLongDatabaseHistory.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.inlong.agent.plugin.utils;
+
+import io.debezium.config.Configuration;
+import io.debezium.relational.history.AbstractDatabaseHistory;
+import io.debezium.relational.history.DatabaseHistoryException;
+import io.debezium.relational.history.DatabaseHistoryListener;
+import io.debezium.relational.history.HistoryRecord;
+import io.debezium.relational.history.HistoryRecordComparator;
+import org.apache.inlong.agent.plugin.message.SchemaRecord;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.function.Consumer;
+
+/**
+ * @description
Review Comment:
resolve
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org