This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new 5f3559a [Bug][Binlog] Fix Bug that multiple sync jobs can connect to
the same canal instance (#6756)
5f3559a is described below
commit 5f3559a94c136358581a9e1ad7f879e94c336201
Author: xy720 <[email protected]>
AuthorDate: Sun Oct 3 12:21:06 2021 +0800
[Bug][Binlog] Fix Bug that multiple sync jobs can connect to the same canal
instance (#6756)
When creating sync jobs, we should ban that different jobs can connect to
the same canal instance,
or else these jobs will compete with each other for the data produced by
the same canal instance,
which may cause data inconsistency.
---
.../org/apache/doris/load/sync/SyncJobManager.java | 16 +++
.../doris/load/sync/canal/CanalDestination.java | 127 +++++++++++++++++++++
.../apache/doris/load/sync/canal/CanalSyncJob.java | 61 ++++------
.../apache/doris/load/sync/SyncJobManagerTest.java | 34 ++++++
4 files changed, 200 insertions(+), 38 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/sync/SyncJobManager.java
b/fe/fe-core/src/main/java/org/apache/doris/load/sync/SyncJobManager.java
index e731e03..1a0d8b8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/sync/SyncJobManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/sync/SyncJobManager.java
@@ -28,6 +28,8 @@ import org.apache.doris.common.UserException;
import org.apache.doris.common.io.Writable;
import org.apache.doris.common.util.LogBuilder;
import org.apache.doris.common.util.LogKey;
+import org.apache.doris.load.sync.canal.CanalDestination;
+import org.apache.doris.load.sync.canal.CanalSyncJob;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -65,6 +67,7 @@ public class SyncJobManager implements Writable {
SyncJob syncJob = SyncJob.fromStmt(jobId, stmt);
writeLock();
try {
+ checkDuplicateRemote(syncJob);
unprotectedAddSyncJob(syncJob);
Catalog.getCurrentCatalog().getEditLog().logCreateSyncJob(syncJob);
} finally {
@@ -78,6 +81,19 @@ public class SyncJobManager implements Writable {
.build());
}
+ private void checkDuplicateRemote(SyncJob syncJob) throws DdlException {
+ if (syncJob.getJobType() == DataSyncJobType.CANAL) {
+ CanalDestination remote = ((CanalSyncJob) syncJob).getRemote();
+ List<SyncJob> unCompletedJobs =
idToSyncJob.values().stream().filter(job -> !job.isCompleted())
+ .collect(Collectors.toList());
+ for (SyncJob job : unCompletedJobs) {
+ if (job instanceof CanalSyncJob && ((CanalSyncJob)
job).getRemote().equals(remote)) {
+ throw new DdlException("Remote Canal instance already
exists. conflict destination: " + remote);
+ }
+ }
+ }
+ }
+
private void unprotectedAddSyncJob(SyncJob syncJob) {
idToSyncJob.put(syncJob.getId(), syncJob);
long dbId = syncJob.getDbId();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/sync/canal/CanalDestination.java
b/fe/fe-core/src/main/java/org/apache/doris/load/sync/canal/CanalDestination.java
new file mode 100644
index 0000000..8aebd0f
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/sync/canal/CanalDestination.java
@@ -0,0 +1,127 @@
+// 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.doris.load.sync.canal;
+
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.google.gson.annotations.SerializedName;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+public class CanalDestination implements Writable {
+
+ @SerializedName(value = "ip")
+ private String ip;
+ @SerializedName(value = "port")
+ private int port;
+ @SerializedName(value = "destination")
+ private String destination;
+
+ public void parse(Map<String, String> properties) throws DdlException {
+ // required binlog properties
+ if (!properties.containsKey(CanalSyncJob.CANAL_SERVER_IP)) {
+ throw new DdlException("Missing " + CanalSyncJob.CANAL_SERVER_IP +
" property in binlog properties");
+ } else {
+ ip = properties.get(CanalSyncJob.CANAL_SERVER_IP);
+ }
+ if (!properties.containsKey(CanalSyncJob.CANAL_SERVER_PORT)) {
+ throw new DdlException("Missing " + CanalSyncJob.CANAL_SERVER_PORT
+ " property in binlog properties");
+ } else {
+ try {
+ port =
Integer.parseInt(properties.get(CanalSyncJob.CANAL_SERVER_PORT));
+ } catch (NumberFormatException e) {
+ throw new DdlException("canal port is not int");
+ }
+ }
+ if (!properties.containsKey(CanalSyncJob.CANAL_DESTINATION)) {
+ throw new DdlException("Missing " + CanalSyncJob.CANAL_DESTINATION
+ " property in binlog properties");
+ } else {
+ destination = properties.get(CanalSyncJob.CANAL_DESTINATION);
+ }
+ }
+
+ public CanalDestination(String ip, int port, String destination) {
+ this.ip = ip;
+ this.port = port;
+ this.destination = destination;
+ }
+
+ public String getIp() {
+ return ip;
+ }
+
+ public void setIp(String ip) {
+ this.ip = ip;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ public String getDestination() {
+ return destination;
+ }
+
+ public void setDestination(String destination) {
+ this.destination = destination;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(ip, port, destination);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == this) return true;
+ if (!(other instanceof CanalDestination)) {
+ return false;
+ }
+ CanalDestination otherDestination = (CanalDestination) other;
+ return ip.equalsIgnoreCase(otherDestination.getIp()) && port ==
otherDestination.getPort() &&
+
destination.equalsIgnoreCase(otherDestination.getDestination());
+ }
+
+ @Override
+ public String toString() {
+ return "CanalDestination [ip=" + ip + ", port=" + port +
+ ", destination=" + destination + "]";
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ String json = GsonUtils.GSON.toJson(this);
+ Text.writeString(out, json);
+ }
+
+ public static CanalDestination read(DataInput in) throws IOException {
+ String json = Text.readString(in);
+ return GsonUtils.GSON.fromJson(json, CanalDestination.class);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/sync/canal/CanalSyncJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/sync/canal/CanalSyncJob.java
index 161c604..7c1b800 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/sync/canal/CanalSyncJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/sync/canal/CanalSyncJob.java
@@ -59,12 +59,8 @@ public class CanalSyncJob extends SyncJob {
protected final static String CANAL_BATCH_SIZE = "canal.batchSize";
protected final static String CANAL_DEBUG = "canal.debug";
- @SerializedName(value = "ip")
- private String ip;
- @SerializedName(value = "port")
- private int port;
- @SerializedName(value = "destination")
- private String destination;
+ @SerializedName(value = "remote")
+ private CanalDestination remote;
@SerializedName(value = "username")
private String username;
@SerializedName(value = "password")
@@ -79,15 +75,16 @@ public class CanalSyncJob extends SyncJob {
public CanalSyncJob(long id, String jobName, long dbId) {
super(id, jobName, dbId);
this.dataSyncJobType = DataSyncJobType.CANAL;
+ this.remote = new CanalDestination("", 0, "");
}
private void init() throws DdlException {
CanalConnector connector = CanalConnectors.newSingleConnector(
- new InetSocketAddress(ip, port), destination, username,
password);
+ new InetSocketAddress(remote.getIp(), remote.getPort()),
remote.getDestination(), username, password);
// create channels
initChannels();
// create client
- client = new SyncCanalClient(this, destination, connector, batchSize,
debug);
+ client = new SyncCanalClient(this, remote.getDestination(), connector,
batchSize, debug);
// register channels into client
client.registerChannels(channels);
}
@@ -129,29 +126,7 @@ public class CanalSyncJob extends SyncJob {
super.checkAndSetBinlogInfo(binlogDesc);
Map<String, String> properties = binlogDesc.getProperties();
- // required binlog properties
- if (!properties.containsKey(CANAL_SERVER_IP)) {
- throw new DdlException("Missing " + CANAL_SERVER_IP + " property
in binlog properties");
- } else {
- ip = properties.get(CANAL_SERVER_IP);
- }
-
- if (!properties.containsKey(CANAL_SERVER_PORT)) {
- throw new DdlException("Missing " + CANAL_SERVER_PORT + " property
in binlog properties");
- } else {
- try {
- port = Integer.parseInt(properties.get(CANAL_SERVER_PORT));
- } catch (NumberFormatException e) {
- throw new DdlException("canal port is not int");
- }
- }
-
- if (!properties.containsKey(CANAL_DESTINATION)) {
- throw new DdlException("Missing " + CANAL_DESTINATION + " property
in binlog properties");
- } else {
- destination = properties.get(CANAL_DESTINATION);
- }
-
+ remote.parse(properties);
if (!properties.containsKey(CANAL_USERNAME)) {
throw new DdlException("Missing " + CANAL_USERNAME + " property in
binlog properties");
} else {
@@ -188,7 +163,13 @@ public class CanalSyncJob extends SyncJob {
@Override
public void execute() throws UserException {
- LOG.info("try to start canal client. Remote ip: {}, remote port: {},
debug: {}", ip, port, debug);
+ LOG.info(new LogBuilder(LogKey.SYNC_JOB, id)
+ .add("remote ip", remote.getIp())
+ .add("remote port", remote.getPort())
+ .add("msg", "Try to start canal client.")
+ .add("debug", debug)
+ .build());
+
// init
if (!isInit()) {
init();
@@ -231,8 +212,8 @@ public class CanalSyncJob extends SyncJob {
public void pause() throws UserException {
unprotectedStopClient(JobState.PAUSED);
LOG.info(new LogBuilder(LogKey.SYNC_JOB, id)
- .add("remote ip", ip)
- .add("remote port", port)
+ .add("remote ip", remote.getIp())
+ .add("remote port", remote.getPort())
.add("msg", "Pause canal sync job.")
.add("debug", debug)
.build());
@@ -242,8 +223,8 @@ public class CanalSyncJob extends SyncJob {
public void resume() throws UserException {
updateState(JobState.PENDING, false);
LOG.info(new LogBuilder(LogKey.SYNC_JOB, id)
- .add("remote ip", ip)
- .add("remote port", port)
+ .add("remote ip", remote.getIp())
+ .add("remote port", remote.getPort())
.add("msg", "Resume canal sync job.")
.add("debug", debug)
.build());
@@ -318,12 +299,16 @@ public class CanalSyncJob extends SyncJob {
@Override
public String getJobConfig() {
StringBuilder sb = new StringBuilder();
- sb.append("address:").append(ip).append(":").append(port).append(",")
- .append("destination:").append(destination).append(",")
+
sb.append("address:").append(remote.getIp()).append(":").append(remote.getPort()).append(",")
+
.append("destination:").append(remote.getDestination()).append(",")
.append("batchSize:").append(batchSize);
return sb.toString();
}
+ public CanalDestination getRemote() {
+ return remote;
+ }
+
@Override
public String toString() {
return "SyncJob [jobId=" + id
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/sync/SyncJobManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/sync/SyncJobManagerTest.java
index 7e35410..8fa080f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/sync/SyncJobManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/sync/SyncJobManagerTest.java
@@ -29,6 +29,7 @@ import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.load.sync.SyncFailMsg.MsgType;
import org.apache.doris.load.sync.SyncJob.JobState;
import org.apache.doris.load.sync.SyncJob.SyncJobUpdateStateInfo;
+import org.apache.doris.load.sync.canal.CanalDestination;
import org.apache.doris.load.sync.canal.CanalSyncJob;
import org.apache.doris.load.sync.canal.SyncCanalClient;
import org.apache.doris.persist.EditLog;
@@ -120,6 +121,39 @@ public class SyncJobManagerTest {
Assert.assertEquals(syncJob1, syncJob2);
}
+ @Test(expected = DdlException.class)
+ public void testCreateDuplicateRemote(@Injectable CreateDataSyncJobStmt
stmt,
+ @Mocked SyncJob syncJob) throws
DdlException {
+ // create two canal jobs
+ CanalSyncJob canalSyncJob1 = new CanalSyncJob(jobId, jobName + "_1",
dbId);
+ CanalSyncJob canalSyncJob2 = new CanalSyncJob(jobId + 1, jobName +
"_2", dbId);
+ new Expectations() {
+ {
+ SyncJob.fromStmt(anyLong, (CreateDataSyncJobStmt) any);
+ returns(canalSyncJob1, canalSyncJob2);
+ }
+ };
+
+ // set same remote destination to two jobs
+ CanalDestination duplicateDes1 = new CanalDestination("dupIp", 1,
"dupDestination");
+ CanalDestination duplicateDes2 = new CanalDestination("dupIp", 1,
"dupDestination");
+ Deencapsulation.setField(canalSyncJob1, "remote", duplicateDes1);
+ Deencapsulation.setField(canalSyncJob2, "remote", duplicateDes2);
+
+ SyncJobManager manager = new SyncJobManager();
+ manager.addDataSyncJob(stmt);
+
+ Map<Long, SyncJob> idToSyncJobs = Deencapsulation.getField(manager,
"idToSyncJob");
+ Assert.assertEquals(1, idToSyncJobs.size());
+ SyncJob syncJob1 = idToSyncJobs.values().iterator().next();
+ Assert.assertEquals(DataSyncJobType.CANAL, syncJob1.getJobType());
+ Assert.assertEquals(duplicateDes1, ((CanalSyncJob)
syncJob1).getRemote());
+
+ // should throw exception
+ manager.addDataSyncJob(stmt);
+ Assert.fail();
+ }
+
@Test
public void testPauseSyncJob(@Injectable PauseSyncJobStmt stmt) {
CanalSyncJob canalSyncJob = new CanalSyncJob(jobId, jobName, dbId);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
