github-forks
/
flink-cdc-connectors
mirror of https://github.com/ververica/flink-cdc-connectors.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[postgres-cdc] Add tests for latest-offset startup strategy (#2527)

Browse Source
pull/2687/head
Hongshun Wang 1 year ago committed by GitHub
parent de45676faa
commit 7da3eaef77
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 169 additions and 113 deletions
Show Stats Download Patch File Download Diff File

268
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/com/ververica/cdc/connectors/postgres/source/PostgresSourceITCase.java
Unescape Escape View File

@ -38,6 +38,8 @@ import org.apache.commons.lang3.StringUtils;
import org.junit.Rule; import org.junit.Rule;
import org.junit.Test; import org.junit.Test;
import org.junit.rules.Timeout; import org.junit.rules.Timeout;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import java.sql.SQLException; import java.sql.SQLException;
import java.util.ArrayList; import java.util.ArrayList;
@ -62,6 +64,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertTrue;
/** IT tests for {@link PostgresSourceBuilder.PostgresIncrementalSource}. */ /** IT tests for {@link PostgresSourceBuilder.PostgresIncrementalSource}. */
@RunWith(Parameterized.class)
public class PostgresSourceITCase extends PostgresTestBase { public class PostgresSourceITCase extends PostgresTestBase {
private static final String DEFAULT_SCAN_STARTUP_MODE = "initial"; private static final String DEFAULT_SCAN_STARTUP_MODE = "initial";
@ -71,6 +74,8 @@ public class PostgresSourceITCase extends PostgresTestBase {
private static final String DB_NAME_PREFIX = "postgres"; private static final String DB_NAME_PREFIX = "postgres";
private static final String SCHEMA_NAME = "customer"; private static final String SCHEMA_NAME = "customer";
private final String scanStartupMode;
@Rule public final Timeout timeoutPerTest = Timeout.seconds(300); @Rule public final Timeout timeoutPerTest = Timeout.seconds(300);
@Rule @Rule
@ -101,6 +106,15 @@ public class PostgresSourceITCase extends PostgresTestBase {
"-U[103, user_3, Hangzhou, 123567891234]", "-U[103, user_3, Hangzhou, 123567891234]",
"+U[103, user_3, Shanghai, 123567891234]"); "+U[103, user_3, Shanghai, 123567891234]");
public PostgresSourceITCase(String scanStartupMode) {
this.scanStartupMode = scanStartupMode;
}
@Parameterized.Parameters(name = "scanStartupMode: {0}")
public static Object[] parameters() {
return new Object[][] {new Object[] {"initial"}, new Object[] {"latest-offset"}};
}
/** Second part stream events, which is made by {@link #makeSecondPartStreamEvents}. */ /** Second part stream events, which is made by {@link #makeSecondPartStreamEvents}. */
private final List<String> secondPartStreamEvents = private final List<String> secondPartStreamEvents =
Arrays.asList( Arrays.asList(
@ -178,82 +192,101 @@ public class PostgresSourceITCase extends PostgresTestBase {
} }
@Test @Test
public void testConsumingTableWithoutPrimaryKey() { public void testConsumingTableWithoutPrimaryKey() throws Exception {
try { if (scanStartupMode == DEFAULT_SCAN_STARTUP_MODE) {
try {
testPostgresParallelSource(
1,
scanStartupMode,
FailoverType.NONE,
FailoverPhase.NEVER,
new String[] {"customers_no_pk"},
RestartStrategies.noRestart());
} catch (Exception e) {
assertTrue(
ExceptionUtils.findThrowableWithMessage(
e,
String.format(
"Incremental snapshot for tables requires primary key, but table %s doesn't have primary key",
SCHEMA_NAME + ".customers_no_pk"))
.isPresent());
}
} else {
testPostgresParallelSource( testPostgresParallelSource(
1, 1,
DEFAULT_SCAN_STARTUP_MODE, scanStartupMode,
FailoverType.NONE, FailoverType.NONE,
FailoverPhase.NEVER, FailoverPhase.NEVER,
new String[] {"customers_no_pk"}, new String[] {"customers_no_pk"},
RestartStrategies.noRestart()); RestartStrategies.noRestart());
} catch (Exception e) {
assertTrue(
ExceptionUtils.findThrowableWithMessage(
e,
String.format(
"Incremental snapshot for tables requires primary key, but table %s doesn't have primary key",
SCHEMA_NAME + ".customers_no_pk"))
.isPresent());
} }
} }
@Test @Test
public void testDebeziumSlotDropOnStop() throws Exception { public void testDebeziumSlotDropOnStop() throws Exception {
String scanStartupMode = DEFAULT_SCAN_STARTUP_MODE; String slotName = getSlotName();
customDatabase.createAndInitialize(); try {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); customDatabase.createAndInitialize();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
env.setParallelism(2);
env.enableCheckpointing(200L); env.setParallelism(2);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0)); env.enableCheckpointing(200L);
String sourceDDL = env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));
format( String sourceDDL =
"CREATE TABLE customers (" format(
+ " id BIGINT NOT NULL," "CREATE TABLE customers ("
+ " name STRING," + " id BIGINT NOT NULL,"
+ " address STRING," + " name STRING,"
+ " phone_number STRING," + " address STRING,"
+ " primary key (id) not enforced" + " phone_number STRING,"
+ ") WITH (" + " primary key (id) not enforced"
+ " 'connector' = 'postgres-cdc'," + ") WITH ("
+ " 'scan.incremental.snapshot.enabled' = 'true'," + " 'connector' = 'postgres-cdc',"
+ " 'hostname' = '%s'," + " 'scan.incremental.snapshot.enabled' = 'true',"
+ " 'port' = '%s'," + " 'hostname' = '%s',"
+ " 'username' = '%s'," + " 'port' = '%s',"
+ " 'password' = '%s'," + " 'username' = '%s',"
+ " 'database-name' = '%s'," + " 'password' = '%s',"
+ " 'schema-name' = '%s'," + " 'database-name' = '%s',"
+ " 'table-name' = '%s'," + " 'schema-name' = '%s',"
+ " 'scan.startup.mode' = '%s'," + " 'table-name' = '%s',"
+ " 'scan.incremental.snapshot.chunk.size' = '100'," + " 'scan.startup.mode' = '%s',"
+ " 'slot.name' = '%s', " + " 'scan.incremental.snapshot.chunk.size' = '100',"
+ " 'debezium.slot.drop.on.stop' = 'true'" + " 'slot.name' = '%s', "
+ ")", + " 'debezium.slot.drop.on.stop' = 'true'"
customDatabase.getHost(), + ")",
customDatabase.getDatabasePort(), customDatabase.getHost(),
customDatabase.getUsername(), customDatabase.getDatabasePort(),
customDatabase.getPassword(), customDatabase.getUsername(),
customDatabase.getDatabaseName(), customDatabase.getPassword(),
SCHEMA_NAME, customDatabase.getDatabaseName(),
"customers", SCHEMA_NAME,
scanStartupMode, "customers",
getSlotName()); scanStartupMode,
tEnv.executeSql(sourceDDL); slotName);
TableResult tableResult = tEnv.executeSql("select * from customers"); tEnv.executeSql(sourceDDL);
TableResult tableResult = tEnv.executeSql("select * from customers");
// first step: check the snapshot data
if (DEFAULT_SCAN_STARTUP_MODE.equals(scanStartupMode)) { // first step: check the snapshot data
checkSnapshotData( if (DEFAULT_SCAN_STARTUP_MODE.equals(scanStartupMode)) {
checkSnapshotData(
tableResult,
FailoverType.JM,
FailoverPhase.STREAM,
new String[] {"customers"});
}
// second step: check the stream data
checkStreamDataWithDDLDuringFailover(
tableResult, FailoverType.JM, FailoverPhase.STREAM, new String[] {"customers"}); tableResult, FailoverType.JM, FailoverPhase.STREAM, new String[] {"customers"});
}
// second step: check the stream data
checkStreamDataWithDDLDuringFailover(
tableResult, FailoverType.JM, FailoverPhase.STREAM, new String[] {"customers"});
tableResult.getJobClient().get().cancel().get(); tableResult.getJobClient().get().cancel().get();
// sleep 1000ms to wait until connections are closed.
Thread.sleep(1000L);
} finally {
customDatabase.removeSlot(slotName);
}
} }
private void testPostgresParallelSource( private void testPostgresParallelSource(
@ -271,7 +304,7 @@ public class PostgresSourceITCase extends PostgresTestBase {
throws Exception { throws Exception {
testPostgresParallelSource( testPostgresParallelSource(
parallelism, parallelism,
DEFAULT_SCAN_STARTUP_MODE, scanStartupMode,
failoverType, failoverType,
failoverPhase, failoverPhase,
captureCustomerTables, captureCustomerTables,
@ -286,56 +319,64 @@ public class PostgresSourceITCase extends PostgresTestBase {
String[] captureCustomerTables, String[] captureCustomerTables,
RestartStrategies.RestartStrategyConfiguration restartStrategyConfiguration) RestartStrategies.RestartStrategyConfiguration restartStrategyConfiguration)
throws Exception { throws Exception {
customDatabase.createAndInitialize(); String slotName = getSlotName();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); try {
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); customDatabase.createAndInitialize();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(parallelism); StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
env.enableCheckpointing(200L);
env.setRestartStrategy(restartStrategyConfiguration); env.setParallelism(parallelism);
String sourceDDL = env.enableCheckpointing(200L);
format( env.setRestartStrategy(restartStrategyConfiguration);
"CREATE TABLE customers (" String sourceDDL =
+ " id BIGINT NOT NULL," format(
+ " name STRING," "CREATE TABLE customers ("
+ " address STRING," + " id BIGINT NOT NULL,"
+ " phone_number STRING," + " name STRING,"
+ " primary key (id) not enforced" + " address STRING,"
+ ") WITH (" + " phone_number STRING,"
+ " 'connector' = 'postgres-cdc'," + " primary key (id) not enforced"
+ " 'scan.incremental.snapshot.enabled' = 'true'," + ") WITH ("
+ " 'hostname' = '%s'," + " 'connector' = 'postgres-cdc',"
+ " 'port' = '%s'," + " 'scan.incremental.snapshot.enabled' = 'true',"
+ " 'username' = '%s'," + " 'hostname' = '%s',"
+ " 'password' = '%s'," + " 'port' = '%s',"
+ " 'database-name' = '%s'," + " 'username' = '%s',"
+ " 'schema-name' = '%s'," + " 'password' = '%s',"
+ " 'table-name' = '%s'," + " 'database-name' = '%s',"
+ " 'scan.startup.mode' = '%s'," + " 'schema-name' = '%s',"
+ " 'scan.incremental.snapshot.chunk.size' = '100'," + " 'table-name' = '%s',"
+ " 'slot.name' = '%s'" + " 'scan.startup.mode' = '%s',"
+ ")", + " 'scan.incremental.snapshot.chunk.size' = '100',"
customDatabase.getHost(), + " 'slot.name' = '%s'"
customDatabase.getDatabasePort(), + ")",
customDatabase.getUsername(), customDatabase.getHost(),
customDatabase.getPassword(), customDatabase.getDatabasePort(),
customDatabase.getDatabaseName(), customDatabase.getUsername(),
SCHEMA_NAME, customDatabase.getPassword(),
getTableNameRegex(captureCustomerTables), customDatabase.getDatabaseName(),
scanStartupMode, SCHEMA_NAME,
getSlotName()); getTableNameRegex(captureCustomerTables),
tEnv.executeSql(sourceDDL); scanStartupMode,
TableResult tableResult = tEnv.executeSql("select * from customers"); slotName);
tEnv.executeSql(sourceDDL);
// first step: check the snapshot data TableResult tableResult = tEnv.executeSql("select * from customers");
if (DEFAULT_SCAN_STARTUP_MODE.equals(scanStartupMode)) {
checkSnapshotData(tableResult, failoverType, failoverPhase, captureCustomerTables); // first step: check the snapshot data
if (DEFAULT_SCAN_STARTUP_MODE.equals(scanStartupMode)) {
checkSnapshotData(tableResult, failoverType, failoverPhase, captureCustomerTables);
}
// second step: check the stream data
checkStreamData(tableResult, failoverType, failoverPhase, captureCustomerTables);
tableResult.getJobClient().get().cancel().get();
// sleep 1000ms to wait until connections are closed.
Thread.sleep(1000L);
} finally {
customDatabase.removeSlot(slotName);
} }
// second step: check the stream data
checkStreamData(tableResult, failoverType, failoverPhase, captureCustomerTables);
tableResult.getJobClient().get().cancel().get();
} }
private void checkSnapshotData( private void checkSnapshotData(
@ -448,6 +489,7 @@ public class PostgresSourceITCase extends PostgresTestBase {
// wait for the stream reading // wait for the stream reading
Thread.sleep(2000L); Thread.sleep(2000L);
// update database during stream fail over period
if (failoverPhase == FailoverPhase.STREAM) { if (failoverPhase == FailoverPhase.STREAM) {
triggerFailover( triggerFailover(
failoverType, failoverType,

13
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/com/ververica/cdc/connectors/postgres/testutils/UniqueDatabase.java
Unescape Escape View File

@ -154,6 +154,19 @@ public class UniqueDatabase {
} }
} }
/**
* Drop slot from database.
*
* @param slotName
*/
public void removeSlot(String slotName) throws SQLException {
String sql = String.format("SELECT pg_drop_replication_slot('%s')", slotName);
try (Connection connection = PostgresTestBase.getJdbcConnection(container, databaseName);
Statement statement = connection.createStatement()) {
statement.execute(sql);
}
}
private String convertSQL(final String sql) { private String convertSQL(final String sql) {
return sql.replace("$DBNAME$", schemaName); return sql.replace("$DBNAME$", schemaName);
} }

1
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/resources/ddl/customer.sql
Unescape Escape View File

@ -108,3 +108,4 @@ VALUES (101,'user_1','Shanghai','123567891234'),
(1018,'user_19','Shanghai','123567891234'), (1018,'user_19','Shanghai','123567891234'),
(1019,'user_20','Shanghai','123567891234'), (1019,'user_20','Shanghai','123567891234'),
(2000,'user_21','Shanghai','123567891234'); (2000,'user_21','Shanghai','123567891234');
ALTER TABLE customers_no_pk REPLICA IDENTITY FULL;
Write Preview
Loading…
Cancel
Save