Skip to content

Commit 4616092

Browse files
whhewhhe
authored
[cdc-source-connector][oceanbase][e2e] Add OceanBase e2e test case (#2521)
This closes #981
1 parent 0c8f5ea commit 4616092

6 files changed

Lines changed: 338 additions & 1 deletion

File tree

‎flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/pom.xml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -356,6 +356,16 @@ under the License.
356356
<outputDirectory>${project.build.directory}/dependencies
357357
</outputDirectory>
358358
</artifactItem>
359+
360+
<artifactItem>
361+
<groupId>com.ververica</groupId>
362+
<artifactId>flink-sql-connector-oceanbase-cdc</artifactId>
363+
<version>${project.version}</version>
364+
<destFileName>oceanbase-cdc-connector.jar</destFileName>
365+
<type>jar</type>
366+
<outputDirectory>${project.build.directory}/dependencies
367+
</outputDirectory>
368+
</artifactItem>
359369
</artifactItems>
360370
</configuration>
361371
</plugin>

‎flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/java/com/ververica/cdc/connectors/tests/OceanBaseE2eITCase.java‎

Lines changed: 259 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,259 @@
1+
/*
2+
* Copyright 2023 Ververica Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.ververica.cdc.connectors.tests;
18+
19+
import com.github.dockerjava.api.DockerClient;
20+
import com.ververica.cdc.connectors.tests.utils.FlinkContainerTestEnvironment;
21+
import com.ververica.cdc.connectors.tests.utils.JdbcProxy;
22+
import com.ververica.cdc.connectors.tests.utils.TestUtils;
23+
import org.junit.AfterClass;
24+
import org.junit.Before;
25+
import org.junit.ClassRule;
26+
import org.junit.Test;
27+
import org.slf4j.Logger;
28+
import org.slf4j.LoggerFactory;
29+
import org.testcontainers.DockerClientFactory;
30+
import org.testcontainers.containers.GenericContainer;
31+
import org.testcontainers.containers.output.Slf4jLogConsumer;
32+
import org.testcontainers.containers.wait.strategy.Wait;
33+
import org.testcontainers.utility.MountableFile;
34+
35+
import java.net.URL;
36+
import java.nio.file.Files;
37+
import java.nio.file.Path;
38+
import java.nio.file.Paths;
39+
import java.sql.Connection;
40+
import java.sql.DriverManager;
41+
import java.sql.SQLException;
42+
import java.sql.Statement;
43+
import java.time.Duration;
44+
import java.util.Arrays;
45+
import java.util.List;
46+
import java.util.regex.Matcher;
47+
import java.util.regex.Pattern;
48+
import java.util.stream.Collectors;
49+
import java.util.stream.Stream;
50+
51+
import static org.junit.Assert.assertNotNull;
52+
53+
/** End-to-end tests for oceanbase-cdc connector uber jar. */
54+
public class OceanBaseE2eITCase extends FlinkContainerTestEnvironment {
55+
56+
private static final Logger LOG = LoggerFactory.getLogger(OceanBaseE2eITCase.class);
57+
58+
private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
59+
60+
private static final Path obCdcJar = TestUtils.getResource("oceanbase-cdc-connector.jar");
61+
private static final Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
62+
63+
// ------------------------------------------------------------------------------------------
64+
// OceanBase container variables
65+
// ------------------------------------------------------------------------------------------
66+
private static final String OB_SERVER_IMAGE = "oceanbase/oceanbase-ce:4.2.0.0";
67+
private static final String OB_LOG_PROXY_IMAGE = "whhe/oblogproxy:1.1.3_4x";
68+
private static final String NETWORK_MODE = "host";
69+
private static final String INTER_CONTAINER_OB_HOST = "host.docker.internal";
70+
private static final String SYS_PASSWORD = "1234567";
71+
private static final String TEST_TENANT = "test";
72+
private static final String TEST_USER = "root@" + TEST_TENANT;
73+
private static final String TEST_PASSWORD = "7654321";
74+
75+
@ClassRule
76+
public static final GenericContainer<?> OB_SERVER =
77+
new GenericContainer<>(OB_SERVER_IMAGE)
78+
.withNetworkMode(NETWORK_MODE)
79+
.withEnv("MODE", "slim")
80+
.withEnv("OB_DATAFILE_SIZE", "1G")
81+
.withEnv("OB_LOG_DISK_SIZE", "4G")
82+
.withEnv("OB_ROOT_PASSWORD", SYS_PASSWORD)
83+
.withEnv("OB_TENANT_NAME", TEST_TENANT)
84+
.withCopyFileToContainer(
85+
MountableFile.forClasspathResource("docker/oceanbase/setup.sql"),
86+
"/root/boot/init.d/init.sql")
87+
.waitingFor(Wait.forLogMessage(".*boot success!.*", 1))
88+
.withStartupTimeout(Duration.ofMinutes(3))
89+
.withLogConsumer(new Slf4jLogConsumer(LOG));
90+
91+
@ClassRule
92+
public static final GenericContainer<?> LOG_PROXY =
93+
new GenericContainer<>(OB_LOG_PROXY_IMAGE)
94+
.withNetworkMode(NETWORK_MODE)
95+
.withEnv("OB_SYS_PASSWORD", SYS_PASSWORD)
96+
.waitingFor(Wait.forLogMessage(".*boot success!.*", 1))
97+
.withStartupTimeout(Duration.ofMinutes(1))
98+
.withLogConsumer(new Slf4jLogConsumer(LOG));
99+
100+
@Before
101+
public void before() {
102+
super.before();
103+
104+
initializeTable("oceanbase_inventory");
105+
}
106+
107+
private Connection getTestConnection(String databaseName) {
108+
try {
109+
Class.forName(MYSQL_DRIVER_CLASS);
110+
return DriverManager.getConnection(
111+
String.format("jdbc:mysql://127.0.0.1:2881/%s?useSSL=false", databaseName),
112+
TEST_USER,
113+
TEST_PASSWORD);
114+
} catch (Exception e) {
115+
throw new RuntimeException("Failed to get test jdbc connection", e);
116+
}
117+
}
118+
119+
@AfterClass
120+
public static void afterClass() {
121+
Stream.of(OB_SERVER, LOG_PROXY).forEach(GenericContainer::stop);
122+
123+
DockerClient client = DockerClientFactory.instance().client();
124+
client.listImagesCmd()
125+
.withImageNameFilter(OB_SERVER_IMAGE)
126+
.exec()
127+
.forEach(image -> client.removeImageCmd(image.getId()).exec());
128+
client.listImagesCmd()
129+
.withImageNameFilter(OB_LOG_PROXY_IMAGE)
130+
.exec()
131+
.forEach(image -> client.removeImageCmd(image.getId()).exec());
132+
}
133+
134+
@Test
135+
public void testOceanBaseCDC() throws Exception {
136+
List<String> sqlLines =
137+
Arrays.asList(
138+
"SET 'execution.checkpointing.interval' = '3s';",
139+
"SET 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true';",
140+
"CREATE TABLE products_source (",
141+
" `id` INT NOT NULL,",
142+
" name STRING,",
143+
" description STRING,",
144+
" weight DECIMAL(10,3),",
145+
" enum_c STRING,",
146+
" json_c STRING,",
147+
" primary key (`id`) not enforced",
148+
") WITH (",
149+
" 'connector' = 'oceanbase-cdc',",
150+
" 'scan.startup.mode' = 'initial',",
151+
" 'username' = '" + TEST_USER + "',",
152+
" 'password' = '" + TEST_PASSWORD + "',",
153+
" 'tenant-name' = '" + TEST_TENANT + "',",
154+
" 'table-list' = 'inventory.products_source',",
155+
" 'hostname' = '" + INTER_CONTAINER_OB_HOST + "',",
156+
" 'port' = '2881',",
157+
" 'jdbc.driver' = '" + MYSQL_DRIVER_CLASS + "',",
158+
" 'logproxy.host' = '" + INTER_CONTAINER_OB_HOST + "',",
159+
" 'logproxy.port' = '2983',",
160+
" 'rootserver-list' = '127.0.0.1:2882:2881',",
161+
" 'working-mode' = 'memory',",
162+
" 'jdbc.properties.useSSL' = 'false'",
163+
");",
164+
"CREATE TABLE ob_products_sink (",
165+
" `id` INT NOT NULL,",
166+
" name STRING,",
167+
" description STRING,",
168+
" weight DECIMAL(10,3),",
169+
" enum_c STRING,",
170+
" json_c STRING,",
171+
" primary key (`id`) not enforced",
172+
") WITH (",
173+
" 'connector' = 'jdbc',",
174+
String.format(
175+
" 'url' = 'jdbc:mysql://%s:3306/%s',",
176+
INTER_CONTAINER_MYSQL_ALIAS,
177+
mysqlInventoryDatabase.getDatabaseName()),
178+
" 'table-name' = 'ob_products_sink',",
179+
" 'username' = '" + MYSQL_TEST_USER + "',",
180+
" 'password' = '" + MYSQL_TEST_PASSWORD + "'",
181+
");",
182+
"INSERT INTO ob_products_sink",
183+
"SELECT * FROM products_source;");
184+
185+
submitSQLJob(sqlLines, obCdcJar, jdbcJar, mysqlDriverJar);
186+
waitUntilJobRunning(Duration.ofSeconds(30));
187+
188+
try (Connection conn = getTestConnection("inventory");
189+
Statement stat = conn.createStatement()) {
190+
stat.execute(
191+
"UPDATE products_source SET description='18oz carpenter hammer' WHERE id=106;");
192+
stat.execute("UPDATE products_source SET weight='5.1' WHERE id=107;");
193+
stat.execute(
194+
"INSERT INTO products_source VALUES (default,'jacket','water resistent white wind breaker',0.2, null, null);");
195+
stat.execute(
196+
"INSERT INTO products_source VALUES (default,'scooter','Big 2-wheel scooter ',5.18, null, null);");
197+
stat.execute(
198+
"UPDATE products_source SET description='new water resistent white wind breaker', weight='0.5' WHERE id=110;");
199+
stat.execute("UPDATE products_source SET weight='5.17' WHERE id=111;");
200+
stat.execute("DELETE FROM products_source WHERE id=111;");
201+
} catch (SQLException e) {
202+
throw new RuntimeException("Update table for CDC failed.", e);
203+
}
204+
205+
String mysqlUrl =
206+
String.format(
207+
"jdbc:mysql://%s:%s/%s",
208+
MYSQL.getHost(),
209+
MYSQL.getDatabasePort(),
210+
mysqlInventoryDatabase.getDatabaseName());
211+
JdbcProxy proxy =
212+
new JdbcProxy(mysqlUrl, MYSQL_TEST_USER, MYSQL_TEST_PASSWORD, MYSQL_DRIVER_CLASS);
213+
List<String> expectResult =
214+
Arrays.asList(
215+
"101,scooter,Small 2-wheel scooter,3.14,red,{\"key1\": \"value1\"}",
216+
"102,car battery,12V car battery,8.1,white,{\"key2\": \"value2\"}",
217+
"103,12-pack drill bits,12-pack of drill bits with sizes ranging from #40 to #3,0.8,red,{\"key3\": \"value3\"}",
218+
"104,hammer,12oz carpenter's hammer,0.75,white,{\"key4\": \"value4\"}",
219+
"105,hammer,14oz carpenter's hammer,0.875,red,{\"k1\": \"v1\", \"k2\": \"v2\"}",
220+
"106,hammer,18oz carpenter hammer,1.0,null,null",
221+
"107,rocks,box of assorted rocks,5.1,null,null",
222+
"108,jacket,water resistent black wind breaker,0.1,null,null",
223+
"109,spare tire,24 inch spare tire,22.2,null,null",
224+
"110,jacket,new water resistent white wind breaker,0.5,null,null");
225+
proxy.checkResultWithTimeout(
226+
expectResult,
227+
"ob_products_sink",
228+
new String[] {"id", "name", "description", "weight", "enum_c", "json_c"},
229+
60000L);
230+
}
231+
232+
protected void initializeTable(String sqlFile) {
233+
final String ddlFile = String.format("ddl/%s.sql", sqlFile);
234+
final URL ddlTestFile = OceanBaseE2eITCase.class.getClassLoader().getResource(ddlFile);
235+
assertNotNull("Cannot locate " + ddlFile, ddlTestFile);
236+
try (Connection connection = getTestConnection("");
237+
Statement statement = connection.createStatement()) {
238+
final List<String> statements =
239+
Arrays.stream(
240+
Files.readAllLines(Paths.get(ddlTestFile.toURI())).stream()
241+
.map(String::trim)
242+
.filter(x -> !x.startsWith("--") && !x.isEmpty())
243+
.map(
244+
x -> {
245+
final Matcher m =
246+
COMMENT_PATTERN.matcher(x);
247+
return m.matches() ? m.group(1) : x;
248+
})
249+
.collect(Collectors.joining("\n"))
250+
.split(";"))
251+
.collect(Collectors.toList());
252+
for (String stmt : statements) {
253+
statement.execute(stmt);
254+
}
255+
} catch (Exception e) {
256+
throw new RuntimeException(e);
257+
}
258+
}
259+
}

‎flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/java/com/ververica/cdc/connectors/tests/utils/FlinkContainerTestEnvironment.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,13 +140,15 @@ public void before() {
140140
new GenericContainer<>(getFlinkDockerImageTag())
141141
.withCommand("jobmanager")
142142
.withNetwork(NETWORK)
143+
.withExtraHost("host.docker.internal", "host-gateway")
143144
.withNetworkAliases(INTER_CONTAINER_JM_ALIAS)
144145
.withExposedPorts(JOB_MANAGER_REST_PORT)
145146
.withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)
146147
.withLogConsumer(new Slf4jLogConsumer(LOG));
147148
taskManager =
148149
new GenericContainer<>(getFlinkDockerImageTag())
149150
.withCommand("taskmanager")
151+
.withExtraHost("host.docker.internal", "host-gateway")
150152
.withNetwork(NETWORK)
151153
.withNetworkAliases(INTER_CONTAINER_TM_ALIAS)
152154
.withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)

‎flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/ddl/mysql_inventory.sql‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,4 +54,13 @@ CREATE TABLE mongodb_products_sink (
5454
name VARCHAR(255) NOT NULL DEFAULT 'flink',
5555
description VARCHAR(512),
5656
weight FLOAT
57-
);
57+
);
58+
59+
CREATE TABLE ob_products_sink (
60+
id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
61+
name VARCHAR(255) NOT NULL DEFAULT 'flink',
62+
description VARCHAR(512),
63+
weight FLOAT,
64+
enum_c VARCHAR(255),
65+
json_c VARCHAR(255)
66+
);

‎flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/ddl/oceanbase_inventory.sql‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
-- Copyright 2023 Ververica Inc.
2+
--
3+
-- Licensed under the Apache License, Version 2.0 (the "License");
4+
-- you may not use this file except in compliance with the License.
5+
-- You may obtain a copy of the License at
6+
-- http://www.apache.org/licenses/LICENSE-2.0
7+
-- Unless required by applicable law or agreed to in writing,
8+
-- software distributed under the License is distributed on an
9+
-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
10+
-- KIND, either express or implied. See the License for the
11+
-- specific language governing permissions and limitations
12+
-- under the License.
13+
14+
-- ----------------------------------------------------------------------------------------------------------------
15+
-- DATABASE: inventory
16+
-- ----------------------------------------------------------------------------------------------------------------
17+
-- Create and populate our products using a single insert with many rows
18+
DROP DATABASE IF EXISTS inventory;
19+
20+
CREATE DATABASE inventory;
21+
22+
USE inventory;
23+
24+
CREATE TABLE products_source (
25+
id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
26+
name VARCHAR(255) NOT NULL DEFAULT 'flink',
27+
description VARCHAR(512),
28+
weight DECIMAL(10, 3),
29+
enum_c enum('red', 'white') default 'red',
30+
json_c JSON
31+
);
32+
ALTER TABLE products_source AUTO_INCREMENT = 101;
33+
34+
INSERT INTO products_source
35+
VALUES (default,"scooter","Small 2-wheel scooter",3.14, 'red', '{"key1": "value1"}'),
36+
(default,"car battery","12V car battery",8.1, 'white', '{"key2": "value2"}'),
37+
(default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8, 'red', '{"key3": "value3"}'),
38+
(default,"hammer","12oz carpenter's hammer",0.75, 'white', '{"key4": "value4"}'),
39+
(default,"hammer","14oz carpenter's hammer",0.875, 'red', '{"k1": "v1", "k2": "v2"}'),
40+
(default,"hammer","16oz carpenter's hammer",1.0, null, null),
41+
(default,"rocks","box of assorted rocks",5.3, null, null),
42+
(default,"jacket","water resistent black wind breaker",0.1, null, null),
43+
(default,"spare tire","24 inch spare tire",22.2, null, null);

‎flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/oceanbase/setup.sql‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
-- Copyright 2023 Ververica Inc.
2+
--
3+
-- Licensed under the Apache License, Version 2.0 (the "License");
4+
-- you may not use this file except in compliance with the License.
5+
-- You may obtain a copy of the License at
6+
-- http://www.apache.org/licenses/LICENSE-2.0
7+
-- Unless required by applicable law or agreed to in writing,
8+
-- software distributed under the License is distributed on an
9+
-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
10+
-- KIND, either express or implied. See the License for the
11+
-- specific language governing permissions and limitations
12+
-- under the License.
13+
14+
ALTER USER root IDENTIFIED BY '7654321';

0 commit comments

Comments
 (0)