[mongodb] Support MongoDB 6.0+ & update testcases
yuxiqian
2 years ago
committed by
Leonard Xu
parent
d3ed1a7714
commit
7f6eaa4b2e
@ -0,0 +1,261 @@
|
||||
/*
|
||||
* Copyright 2023 Ververica Inc.
|
||||
*
|
||||
* Licensed 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 com.ververica.cdc.connectors.mongodb;
|
||||
|
||||
import com.github.dockerjava.api.command.InspectContainerResponse;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.testcontainers.containers.GenericContainer;
|
||||
import org.testcontainers.containers.Network;
|
||||
import org.testcontainers.containers.wait.strategy.Wait;
|
||||
import org.testcontainers.containers.wait.strategy.WaitStrategy;
|
||||
import org.testcontainers.images.builder.ImageFromDockerfile;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.URL;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.Random;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
|
||||
/** Mongodb test container. */
|
||||
public class LegacyMongoDBContainer extends GenericContainer<LegacyMongoDBContainer> {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(LegacyMongoDBContainer.class);
|
||||
|
||||
private static final String DOCKER_IMAGE_NAME = "mongo:5.0.2";
|
||||
|
||||
public static final int MONGODB_PORT = 27017;
|
||||
|
||||
public static final String MONGO_SUPER_USER = "superuser";
|
||||
|
||||
public static final String MONGO_SUPER_PASSWORD = "superpw";
|
||||
|
||||
public static final String FLINK_USER = "flinkuser";
|
||||
|
||||
public static final String FLINK_USER_PASSWORD = "a1?~!@#$%^&*(){}[]<>.,+_-=/|:;";
|
||||
|
||||
private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)//.*$");
|
||||
|
||||
private final ShardingClusterRole clusterRole;
|
||||
|
||||
public LegacyMongoDBContainer(Network network) {
|
||||
this(network, ShardingClusterRole.NONE);
|
||||
}
|
||||
|
||||
public LegacyMongoDBContainer(Network network, ShardingClusterRole clusterRole) {
|
||||
super(
|
||||
new ImageFromDockerfile()
|
||||
.withFileFromClasspath("random.key", "docker/mongodb/random.key")
|
||||
.withFileFromClasspath("setup.js", "docker/mongodb/setup.js")
|
||||
.withDockerfileFromBuilder(
|
||||
builder ->
|
||||
builder.from(DOCKER_IMAGE_NAME)
|
||||
.copy(
|
||||
"setup.js",
|
||||
"/docker-entrypoint-initdb.d/setup.js")
|
||||
.copy("random.key", "/data/keyfile/random.key")
|
||||
.run("chown mongodb /data/keyfile/random.key")
|
||||
.run("chmod 400 /data/keyfile/random.key")
|
||||
.env("MONGO_INITDB_ROOT_USERNAME", MONGO_SUPER_USER)
|
||||
.env(
|
||||
"MONGO_INITDB_ROOT_PASSWORD",
|
||||
MONGO_SUPER_PASSWORD)
|
||||
.env("MONGO_INITDB_DATABASE", "admin")
|
||||
.build()));
|
||||
this.clusterRole = clusterRole;
|
||||
|
||||
withNetwork(network);
|
||||
withNetworkAliases(clusterRole.hostname);
|
||||
withExposedPorts(MONGODB_PORT);
|
||||
withCommand(ShardingClusterRole.startupCommand(clusterRole));
|
||||
waitingFor(clusterRole.waitStrategy);
|
||||
}
|
||||
|
||||
public String getConnectionString(String username, String password) {
|
||||
return String.format(
|
||||
"mongodb://%s:%s@%s:%d",
|
||||
username, password, getContainerIpAddress(), getMappedPort(MONGODB_PORT));
|
||||
}
|
||||
|
||||
public String getHostAndPort() {
|
||||
return String.format("%s:%s", getContainerIpAddress(), getMappedPort(MONGODB_PORT));
|
||||
}
|
||||
|
||||
public void executeCommand(String command) {
|
||||
try {
|
||||
LOG.info("Executing mongo command: {}", command);
|
||||
ExecResult execResult =
|
||||
execInContainer(
|
||||
"mongo",
|
||||
"-u",
|
||||
MONGO_SUPER_USER,
|
||||
"-p",
|
||||
MONGO_SUPER_PASSWORD,
|
||||
"--eval",
|
||||
command);
|
||||
LOG.info(execResult.getStdout());
|
||||
if (execResult.getExitCode() != 0) {
|
||||
throw new IllegalStateException(
|
||||
"Execute mongo command failed " + execResult.getStdout());
|
||||
}
|
||||
} catch (InterruptedException | IOException e) {
|
||||
throw new IllegalStateException("Execute mongo command failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void containerIsStarted(InspectContainerResponse containerInfo) {
|
||||
LOG.info("Preparing a MongoDB Container with sharding cluster role {}...", clusterRole);
|
||||
if (clusterRole != ShardingClusterRole.ROUTER) {
|
||||
initReplicaSet();
|
||||
} else {
|
||||
initShard();
|
||||
}
|
||||
}
|
||||
|
||||
protected void initReplicaSet() {
|
||||
LOG.info("Initializing a single node replica set...");
|
||||
executeCommand(
|
||||
String.format(
|
||||
"rs.initiate({ _id : '%s', configsvr: %s, members: [{ _id: 0, host: '%s:%d'}]})",
|
||||
clusterRole.replicaSetName,
|
||||
clusterRole == ShardingClusterRole.CONFIG,
|
||||
clusterRole.hostname,
|
||||
MONGODB_PORT));
|
||||
|
||||
LOG.info("Waiting for single node replica set initialized...");
|
||||
executeCommand(
|
||||
String.format(
|
||||
"var attempt = 0; "
|
||||
+ "while"
|
||||
+ "(%s) "
|
||||
+ "{ "
|
||||
+ "if (attempt > %d) {quit(1);} "
|
||||
+ "print('%s ' + attempt); sleep(100); attempt++; "
|
||||
+ " }",
|
||||
"db.runCommand( { isMaster: 1 } ).ismaster==false",
|
||||
60,
|
||||
"An attempt to await for a single node replica set initialization:"));
|
||||
}
|
||||
|
||||
protected void initShard() {
|
||||
LOG.info("Initializing a sharded cluster...");
|
||||
// decrease chunk size from default 64mb to 1mb to make splitter test easier.
|
||||
executeCommand(
|
||||
"db.getSiblingDB('config').settings.updateOne(\n"
|
||||
+ " { _id: \"chunksize\" },\n"
|
||||
+ " { $set: { _id: \"chunksize\", value: 1 } },\n"
|
||||
+ " { upsert: true }\n"
|
||||
+ ");");
|
||||
executeCommand(
|
||||
String.format(
|
||||
"sh.addShard('%s/%s:%d')",
|
||||
ShardingClusterRole.SHARD.replicaSetName,
|
||||
ShardingClusterRole.SHARD.hostname,
|
||||
MONGODB_PORT));
|
||||
}
|
||||
|
||||
/** Executes a mongo command file in separate database. */
|
||||
public String executeCommandFileInSeparateDatabase(String fileNameIgnoreSuffix) {
|
||||
return executeCommandFileInDatabase(
|
||||
fileNameIgnoreSuffix,
|
||||
fileNameIgnoreSuffix + "_" + Integer.toUnsignedString(new Random().nextInt(), 36));
|
||||
}
|
||||
|
||||
/** Executes a mongo command file, specify a database name. */
|
||||
public String executeCommandFileInDatabase(String fileNameIgnoreSuffix, String databaseName) {
|
||||
final String dbName = databaseName != null ? databaseName : fileNameIgnoreSuffix;
|
||||
final String ddlFile = String.format("ddl/%s.js", fileNameIgnoreSuffix);
|
||||
final URL ddlTestFile = LegacyMongoDBContainer.class.getClassLoader().getResource(ddlFile);
|
||||
assertNotNull("Cannot locate " + ddlFile, ddlTestFile);
|
||||
|
||||
try {
|
||||
// use database;
|
||||
String command0 = String.format("db = db.getSiblingDB('%s');\n", dbName);
|
||||
String command1 =
|
||||
Files.readAllLines(Paths.get(ddlTestFile.toURI())).stream()
|
||||
.filter(x -> StringUtils.isNotBlank(x) && !x.trim().startsWith("//"))
|
||||
.map(
|
||||
x -> {
|
||||
final Matcher m = COMMENT_PATTERN.matcher(x);
|
||||
return m.matches() ? m.group(1) : x;
|
||||
})
|
||||
.collect(Collectors.joining("\n"));
|
||||
|
||||
executeCommand(command0 + command1);
|
||||
|
||||
return dbName;
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/** A MongoDB sharded cluster roles. */
|
||||
public enum ShardingClusterRole {
|
||||
// Config servers store metadata and configuration settings for the cluster.
|
||||
CONFIG("config0", "rs0-config", Wait.forLogMessage(".*[Ww]aiting for connections.*", 2)),
|
||||
|
||||
// Each shard contains a subset of the sharded data. Each shard can be deployed as a replica
|
||||
// set.
|
||||
SHARD("shard0", "rs0-shard", Wait.forLogMessage(".*[Ww]aiting for connections.*", 2)),
|
||||
|
||||
// The mongos acts as a query router, providing an interface between client applications and
|
||||
// the sharded cluster.
|
||||
ROUTER("router0", null, Wait.forLogMessage(".*[Ww]aiting for connections.*", 1)),
|
||||
|
||||
// None sharded cluster.
|
||||
NONE("mongo0", "rs0", Wait.forLogMessage(".*Replication has not yet been configured.*", 1));
|
||||
|
||||
private final String hostname;
|
||||
private final String replicaSetName;
|
||||
private final WaitStrategy waitStrategy;
|
||||
|
||||
ShardingClusterRole(String hostname, String replicaSetName, WaitStrategy waitStrategy) {
|
||||
this.hostname = hostname;
|
||||
this.replicaSetName = replicaSetName;
|
||||
this.waitStrategy = waitStrategy;
|
||||
}
|
||||
|
||||
public static String startupCommand(ShardingClusterRole clusterRole) {
|
||||
switch (clusterRole) {
|
||||
case CONFIG:
|
||||
return String.format(
|
||||
"mongod --configsvr --port %d --replSet %s --keyFile /data/keyfile/random.key",
|
||||
MONGODB_PORT, clusterRole.replicaSetName);
|
||||
case SHARD:
|
||||
return String.format(
|
||||
"mongod --shardsvr --port %d --replSet %s --keyFile /data/keyfile/random.key",
|
||||
MONGODB_PORT, clusterRole.replicaSetName);
|
||||
case ROUTER:
|
||||
return String.format(
|
||||
"mongos --configdb %s/%s:%d --bind_ip_all --keyFile /data/keyfile/random.key",
|
||||
CONFIG.replicaSetName, CONFIG.hostname, MONGODB_PORT);
|
||||
case NONE:
|
||||
default:
|
||||
return String.format(
|
||||
"mongod --port %d --replSet %s --keyFile /data/keyfile/random.key",
|
||||
MONGODB_PORT, NONE.replicaSetName);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,116 @@
|
||||
/*
|
||||
* Copyright 2023 Ververica Inc.
|
||||
*
|
||||
* Licensed 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 com.ververica.cdc.connectors.mongodb;
|
||||
|
||||
import org.apache.flink.runtime.minicluster.RpcServiceSharing;
|
||||
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
|
||||
import org.apache.flink.test.util.MiniClusterWithClientResource;
|
||||
import org.apache.flink.util.TestLogger;
|
||||
|
||||
import com.mongodb.ConnectionString;
|
||||
import com.mongodb.MongoClientSettings;
|
||||
import com.mongodb.client.MongoClient;
|
||||
import com.mongodb.client.MongoClients;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Rule;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.testcontainers.containers.Network;
|
||||
import org.testcontainers.containers.output.Slf4jLogConsumer;
|
||||
import org.testcontainers.lifecycle.Startables;
|
||||
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import static com.ververica.cdc.connectors.mongodb.LegacyMongoDBContainer.MONGO_SUPER_PASSWORD;
|
||||
import static com.ververica.cdc.connectors.mongodb.LegacyMongoDBContainer.MONGO_SUPER_USER;
|
||||
|
||||
/** Basic class for testing {@link MongoDBSource}. */
|
||||
public abstract class LegacyMongoDBSourceTestBase extends TestLogger {
|
||||
|
||||
protected static final Logger LOG = LoggerFactory.getLogger(LegacyMongoDBSourceTestBase.class);
|
||||
protected static final int DEFAULT_PARALLELISM = 4;
|
||||
|
||||
@Rule
|
||||
public final MiniClusterWithClientResource miniClusterResource =
|
||||
new MiniClusterWithClientResource(
|
||||
new MiniClusterResourceConfiguration.Builder()
|
||||
.setNumberTaskManagers(1)
|
||||
.setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
|
||||
.setRpcServiceSharing(RpcServiceSharing.DEDICATED)
|
||||
.withHaLeadershipControl()
|
||||
.build());
|
||||
|
||||
@ClassRule public static final Network NETWORK = Network.newNetwork();
|
||||
|
||||
@ClassRule
|
||||
public static final LegacyMongoDBContainer CONFIG =
|
||||
new LegacyMongoDBContainer(NETWORK, LegacyMongoDBContainer.ShardingClusterRole.CONFIG)
|
||||
.withLogConsumer(new Slf4jLogConsumer(LOG));
|
||||
|
||||
@ClassRule
|
||||
public static final LegacyMongoDBContainer SHARD =
|
||||
new LegacyMongoDBContainer(NETWORK, LegacyMongoDBContainer.ShardingClusterRole.SHARD)
|
||||
.dependsOn(CONFIG)
|
||||
.withLogConsumer(new Slf4jLogConsumer(LOG));
|
||||
|
||||
@ClassRule
|
||||
public static final LegacyMongoDBContainer ROUTER =
|
||||
new LegacyMongoDBContainer(NETWORK, LegacyMongoDBContainer.ShardingClusterRole.ROUTER)
|
||||
.dependsOn(SHARD)
|
||||
.withLogConsumer(new Slf4jLogConsumer(LOG));
|
||||
|
||||
protected static MongoClient mongodbClient;
|
||||
|
||||
@BeforeClass
|
||||
public static void startContainers() {
|
||||
LOG.info("Starting containers...");
|
||||
Startables.deepStart(Stream.of(CONFIG)).join();
|
||||
Startables.deepStart(Stream.of(SHARD)).join();
|
||||
Startables.deepStart(Stream.of(ROUTER)).join();
|
||||
initialClient();
|
||||
LOG.info("Containers are started.");
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void closeContainers() {
|
||||
if (mongodbClient != null) {
|
||||
mongodbClient.close();
|
||||
}
|
||||
if (ROUTER != null) {
|
||||
ROUTER.close();
|
||||
}
|
||||
if (SHARD != null) {
|
||||
SHARD.close();
|
||||
}
|
||||
if (CONFIG != null) {
|
||||
CONFIG.close();
|
||||
}
|
||||
}
|
||||
|
||||
private static void initialClient() {
|
||||
MongoClientSettings settings =
|
||||
MongoClientSettings.builder()
|
||||
.applyConnectionString(
|
||||
new ConnectionString(
|
||||
ROUTER.getConnectionString(
|
||||
MONGO_SUPER_USER, MONGO_SUPER_PASSWORD)))
|
||||
.build();
|
||||
mongodbClient = MongoClients.create(settings);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,384 @@
|
||||
/*
|
||||
* Copyright 2023 Ververica Inc.
|
||||
*
|
||||
* Licensed 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 com.ververica.cdc.connectors.mongodb.source;
|
||||
|
||||
import org.apache.flink.api.common.JobID;
|
||||
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
|
||||
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
|
||||
import org.apache.flink.table.api.TableResult;
|
||||
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
|
||||
import org.apache.flink.types.Row;
|
||||
import org.apache.flink.util.CloseableIterator;
|
||||
|
||||
import com.mongodb.client.MongoCollection;
|
||||
import com.mongodb.client.MongoDatabase;
|
||||
import com.mongodb.client.model.Filters;
|
||||
import com.mongodb.client.model.Updates;
|
||||
import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceConfig;
|
||||
import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceConfigFactory;
|
||||
import com.ververica.cdc.connectors.mongodb.source.utils.MongoUtils;
|
||||
import com.ververica.cdc.connectors.mongodb.utils.MongoDBTestUtils;
|
||||
import org.bson.Document;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.Timeout;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.runners.Parameterized;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Random;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static com.ververica.cdc.connectors.mongodb.utils.MongoDBAssertUtils.assertEqualsInAnyOrder;
|
||||
import static com.ververica.cdc.connectors.mongodb.utils.MongoDBContainer.FLINK_USER;
|
||||
import static com.ververica.cdc.connectors.mongodb.utils.MongoDBContainer.FLINK_USER_PASSWORD;
|
||||
import static com.ververica.cdc.connectors.mongodb.utils.MongoDBTestUtils.fetchRows;
|
||||
import static com.ververica.cdc.connectors.mongodb.utils.MongoDBTestUtils.triggerFailover;
|
||||
import static org.apache.flink.util.Preconditions.checkState;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
/** Integration tests for MongoDB full document before change info. */
|
||||
@RunWith(Parameterized.class)
|
||||
public class MongoDBFullChangelogITCase extends MongoDBSourceTestBase {
|
||||
|
||||
@Rule public final Timeout timeoutPerTest = Timeout.seconds(300);
|
||||
|
||||
private final boolean parallelismSnapshot;
|
||||
|
||||
public MongoDBFullChangelogITCase(boolean parallelismSnapshot) {
|
||||
this.parallelismSnapshot = parallelismSnapshot;
|
||||
}
|
||||
|
||||
@Parameterized.Parameters(name = "parallelismSnapshot: {0}")
|
||||
public static Object[] parameters() {
|
||||
return new Object[][] {new Object[] {false}, new Object[] {true}};
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetMongoDBVersion() {
|
||||
MongoDBSourceConfig config =
|
||||
new MongoDBSourceConfigFactory()
|
||||
.hosts(CONTAINER.getHostAndPort())
|
||||
.splitSizeMB(1)
|
||||
.pollAwaitTimeMillis(500)
|
||||
.create(0);
|
||||
|
||||
assertEquals(MongoUtils.getMongoVersion(config), "6.0.6");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReadSingleCollectionWithSingleParallelism() throws Exception {
|
||||
testMongoDBParallelSource(
|
||||
1,
|
||||
MongoDBTestUtils.FailoverType.NONE,
|
||||
MongoDBTestUtils.FailoverPhase.NEVER,
|
||||
new String[] {"customers"});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReadSingleCollectionWithMultipleParallelism() throws Exception {
|
||||
testMongoDBParallelSource(
|
||||
4,
|
||||
MongoDBTestUtils.FailoverType.NONE,
|
||||
MongoDBTestUtils.FailoverPhase.NEVER,
|
||||
new String[] {"customers"});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReadMultipleCollectionWithSingleParallelism() throws Exception {
|
||||
testMongoDBParallelSource(
|
||||
1,
|
||||
MongoDBTestUtils.FailoverType.NONE,
|
||||
MongoDBTestUtils.FailoverPhase.NEVER,
|
||||
new String[] {"customers", "customers_1"});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReadMultipleCollectionWithMultipleParallelism() throws Exception {
|
||||
testMongoDBParallelSource(
|
||||
4,
|
||||
MongoDBTestUtils.FailoverType.NONE,
|
||||
MongoDBTestUtils.FailoverPhase.NEVER,
|
||||
new String[] {"customers", "customers_1"});
|
||||
}
|
||||
|
||||
// Failover tests
|
||||
@Test
|
||||
public void testTaskManagerFailoverInSnapshotPhase() throws Exception {
|
||||
if (!parallelismSnapshot) {
|
||||
return;
|
||||
}
|
||||
testMongoDBParallelSource(
|
||||
MongoDBTestUtils.FailoverType.TM,
|
||||
MongoDBTestUtils.FailoverPhase.SNAPSHOT,
|
||||
new String[] {"customers", "customers_1"});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTaskManagerFailoverInStreamPhase() throws Exception {
|
||||
if (!parallelismSnapshot) {
|
||||
return;
|
||||
}
|
||||
testMongoDBParallelSource(
|
||||
MongoDBTestUtils.FailoverType.TM,
|
||||
MongoDBTestUtils.FailoverPhase.STREAM,
|
||||
new String[] {"customers", "customers_1"});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testJobManagerFailoverInSnapshotPhase() throws Exception {
|
||||
if (!parallelismSnapshot) {
|
||||
return;
|
||||
}
|
||||
testMongoDBParallelSource(
|
||||
MongoDBTestUtils.FailoverType.JM,
|
||||
MongoDBTestUtils.FailoverPhase.SNAPSHOT,
|
||||
new String[] {"customers", "customers_1"});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testJobManagerFailoverInStreamPhase() throws Exception {
|
||||
if (!parallelismSnapshot) {
|
||||
return;
|
||||
}
|
||||
testMongoDBParallelSource(
|
||||
MongoDBTestUtils.FailoverType.JM,
|
||||
MongoDBTestUtils.FailoverPhase.STREAM,
|
||||
new String[] {"customers", "customers_1"});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTaskManagerFailoverSingleParallelism() throws Exception {
|
||||
if (!parallelismSnapshot) {
|
||||
return;
|
||||
}
|
||||
testMongoDBParallelSource(
|
||||
1,
|
||||
MongoDBTestUtils.FailoverType.TM,
|
||||
MongoDBTestUtils.FailoverPhase.SNAPSHOT,
|
||||
new String[] {"customers"});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testJobManagerFailoverSingleParallelism() throws Exception {
|
||||
if (!parallelismSnapshot) {
|
||||
return;
|
||||
}
|
||||
testMongoDBParallelSource(
|
||||
1,
|
||||
MongoDBTestUtils.FailoverType.JM,
|
||||
MongoDBTestUtils.FailoverPhase.SNAPSHOT,
|
||||
new String[] {"customers"});
|
||||
}
|
||||
|
||||
private void testMongoDBParallelSource(
|
||||
MongoDBTestUtils.FailoverType failoverType,
|
||||
MongoDBTestUtils.FailoverPhase failoverPhase,
|
||||
String[] captureCustomerCollections)
|
||||
throws Exception {
|
||||
testMongoDBParallelSource(
|
||||
DEFAULT_PARALLELISM, failoverType, failoverPhase, captureCustomerCollections);
|
||||
}
|
||||
|
||||
private void testMongoDBParallelSource(
|
||||
int parallelism,
|
||||
MongoDBTestUtils.FailoverType failoverType,
|
||||
MongoDBTestUtils.FailoverPhase failoverPhase,
|
||||
String[] captureCustomerCollections)
|
||||
throws Exception {
|
||||
|
||||
String customerDatabase =
|
||||
"customer_" + Integer.toUnsignedString(new Random().nextInt(), 36);
|
||||
|
||||
// A - enable system-level fulldoc pre & post image feature
|
||||
CONTAINER.executeCommand(
|
||||
"use admin; db.runCommand({ setClusterParameter: { changeStreamOptions: { preAndPostImages: { expireAfterSeconds: 'off' } } } })");
|
||||
|
||||
// B - enable collection-level fulldoc pre & post image for change capture collection
|
||||
for (String collectionName : captureCustomerCollections) {
|
||||
CONTAINER.executeCommandInDatabase(
|
||||
String.format(
|
||||
"db.createCollection('%s'); db.runCommand({ collMod: '%s', changeStreamPreAndPostImages: { enabled: true } })",
|
||||
collectionName, collectionName),
|
||||
customerDatabase);
|
||||
}
|
||||
|
||||
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
|
||||
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
|
||||
|
||||
env.setParallelism(parallelism);
|
||||
env.enableCheckpointing(200L);
|
||||
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));
|
||||
|
||||
String sourceDDL =
|
||||
String.format(
|
||||
"CREATE TABLE customers ("
|
||||
+ " _id STRING NOT NULL,"
|
||||
+ " cid BIGINT NOT NULL,"
|
||||
+ " name STRING,"
|
||||
+ " address STRING,"
|
||||
+ " phone_number STRING,"
|
||||
+ " primary key (_id) not enforced"
|
||||
+ ") WITH ("
|
||||
+ " 'connector' = 'mongodb-cdc',"
|
||||
+ " 'scan.incremental.snapshot.enabled' = '%s',"
|
||||
+ " 'hosts' = '%s',"
|
||||
+ " 'username' = '%s',"
|
||||
+ " 'password' = '%s',"
|
||||
+ " 'database' = '%s',"
|
||||
+ " 'collection' = '%s',"
|
||||
+ " 'heartbeat.interval.ms' = '500',"
|
||||
+ " 'scan.full-changelog' = 'true'"
|
||||
+ ")",
|
||||
parallelismSnapshot ? "true" : "false",
|
||||
CONTAINER.getHostAndPort(),
|
||||
FLINK_USER,
|
||||
FLINK_USER_PASSWORD,
|
||||
customerDatabase,
|
||||
getCollectionNameRegex(customerDatabase, captureCustomerCollections));
|
||||
|
||||
CONTAINER.executeCommandFileInDatabase("customer", customerDatabase);
|
||||
|
||||
// first step: check the snapshot data
|
||||
String[] snapshotForSingleTable =
|
||||
new String[] {
|
||||
"+I[101, user_1, Shanghai, 123567891234]",
|
||||
"+I[102, user_2, Shanghai, 123567891234]",
|
||||
"+I[103, user_3, Shanghai, 123567891234]",
|
||||
"+I[109, user_4, Shanghai, 123567891234]",
|
||||
"+I[110, user_5, Shanghai, 123567891234]",
|
||||
"+I[111, user_6, Shanghai, 123567891234]",
|
||||
"+I[118, user_7, Shanghai, 123567891234]",
|
||||
"+I[121, user_8, Shanghai, 123567891234]",
|
||||
"+I[123, user_9, Shanghai, 123567891234]",
|
||||
"+I[1009, user_10, Shanghai, 123567891234]",
|
||||
"+I[1010, user_11, Shanghai, 123567891234]",
|
||||
"+I[1011, user_12, Shanghai, 123567891234]",
|
||||
"+I[1012, user_13, Shanghai, 123567891234]",
|
||||
"+I[1013, user_14, Shanghai, 123567891234]",
|
||||
"+I[1014, user_15, Shanghai, 123567891234]",
|
||||
"+I[1015, user_16, Shanghai, 123567891234]",
|
||||
"+I[1016, user_17, Shanghai, 123567891234]",
|
||||
"+I[1017, user_18, Shanghai, 123567891234]",
|
||||
"+I[1018, user_19, Shanghai, 123567891234]",
|
||||
"+I[1019, user_20, Shanghai, 123567891234]",
|
||||
"+I[2000, user_21, Shanghai, 123567891234]"
|
||||
};
|
||||
tEnv.executeSql(sourceDDL);
|
||||
TableResult tableResult =
|
||||
tEnv.executeSql("select cid, name, address, phone_number from customers");
|
||||
CloseableIterator<Row> iterator = tableResult.collect();
|
||||
JobID jobId = tableResult.getJobClient().get().getJobID();
|
||||
List<String> expectedSnapshotData = new ArrayList<>();
|
||||
for (int i = 0; i < captureCustomerCollections.length; i++) {
|
||||
expectedSnapshotData.addAll(Arrays.asList(snapshotForSingleTable));
|
||||
}
|
||||
|
||||
// trigger failover after some snapshot splits read finished
|
||||
if (failoverPhase == MongoDBTestUtils.FailoverPhase.SNAPSHOT && iterator.hasNext()) {
|
||||
triggerFailover(
|
||||
failoverType, jobId, miniClusterResource.getMiniCluster(), () -> sleepMs(100));
|
||||
}
|
||||
|
||||
assertEqualsInAnyOrder(
|
||||
expectedSnapshotData, fetchRows(iterator, expectedSnapshotData.size()));
|
||||
|
||||
// second step: check the change stream data
|
||||
for (String collectionName : captureCustomerCollections) {
|
||||
makeFirstPartChangeStreamEvents(
|
||||
mongodbClient.getDatabase(customerDatabase), collectionName);
|
||||
}
|
||||
if (failoverPhase == MongoDBTestUtils.FailoverPhase.STREAM) {
|
||||
triggerFailover(
|
||||
failoverType, jobId, miniClusterResource.getMiniCluster(), () -> sleepMs(200));
|
||||
}
|
||||
for (String collectionName : captureCustomerCollections) {
|
||||
makeSecondPartChangeStreamEvents(
|
||||
mongodbClient.getDatabase(customerDatabase), collectionName);
|
||||
}
|
||||
|
||||
String[] changeEventsForSingleTable =
|
||||
new String[] {
|
||||
"-U[101, user_1, Shanghai, 123567891234]",
|
||||
"+U[101, user_1, Hangzhou, 123567891234]",
|
||||
"-D[102, user_2, Shanghai, 123567891234]",
|
||||
"+I[102, user_2, Shanghai, 123567891234]",
|
||||
"-U[103, user_3, Shanghai, 123567891234]",
|
||||
"+U[103, user_3, Hangzhou, 123567891234]",
|
||||
"-U[1010, user_11, Shanghai, 123567891234]",
|
||||
"+U[1010, user_11, Hangzhou, 123567891234]",
|
||||
"+I[2001, user_22, Shanghai, 123567891234]",
|
||||
"+I[2002, user_23, Shanghai, 123567891234]",
|
||||
"+I[2003, user_24, Shanghai, 123567891234]"
|
||||
};
|
||||
List<String> expectedChangeStreamData = new ArrayList<>();
|
||||
for (int i = 0; i < captureCustomerCollections.length; i++) {
|
||||
expectedChangeStreamData.addAll(Arrays.asList(changeEventsForSingleTable));
|
||||
}
|
||||
List<String> actualChangeStreamData = fetchRows(iterator, expectedChangeStreamData.size());
|
||||
assertEqualsInAnyOrder(expectedChangeStreamData, actualChangeStreamData);
|
||||
tableResult.getJobClient().get().cancel().get();
|
||||
}
|
||||
|
||||
private String getCollectionNameRegex(String database, String[] captureCustomerCollections) {
|
||||
checkState(captureCustomerCollections.length > 0);
|
||||
if (captureCustomerCollections.length == 1) {
|
||||
return captureCustomerCollections[0];
|
||||
} else {
|
||||
// pattern that matches multiple collections
|
||||
return Arrays.stream(captureCustomerCollections)
|
||||
.map(coll -> "^(" + database + "." + coll + ")$")
|
||||
.collect(Collectors.joining("|"));
|
||||
}
|
||||
}
|
||||
|
||||
private void sleepMs(long millis) {
|
||||
try {
|
||||
Thread.sleep(millis);
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
}
|
||||
|
||||
private void makeFirstPartChangeStreamEvents(MongoDatabase mongoDatabase, String collection) {
|
||||
MongoCollection<Document> mongoCollection = mongoDatabase.getCollection(collection);
|
||||
mongoCollection.updateOne(Filters.eq("cid", 101L), Updates.set("address", "Hangzhou"));
|
||||
mongoCollection.deleteOne(Filters.eq("cid", 102L));
|
||||
mongoCollection.insertOne(customerDocOf(102L, "user_2", "Shanghai", "123567891234"));
|
||||
mongoCollection.updateOne(Filters.eq("cid", 103L), Updates.set("address", "Hangzhou"));
|
||||
}
|
||||
|
||||
private void makeSecondPartChangeStreamEvents(MongoDatabase mongoDatabase, String collection) {
|
||||
MongoCollection<Document> mongoCollection = mongoDatabase.getCollection(collection);
|
||||
mongoCollection.updateOne(Filters.eq("cid", 1010L), Updates.set("address", "Hangzhou"));
|
||||
mongoCollection.insertMany(
|
||||
Arrays.asList(
|
||||
customerDocOf(2001L, "user_22", "Shanghai", "123567891234"),
|
||||
customerDocOf(2002L, "user_23", "Shanghai", "123567891234"),
|
||||
customerDocOf(2003L, "user_24", "Shanghai", "123567891234")));
|
||||
}
|
||||
|
||||
private Document customerDocOf(Long cid, String name, String address, String phoneNumber) {
|
||||
Document document = new Document();
|
||||
document.put("cid", cid);
|
||||
document.put("name", name);
|
||||
document.put("address", address);
|
||||
document.put("phone_number", phoneNumber);
|
||||
return document;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue