|
20 | 20 | import org.apache.flink.runtime.minicluster.RpcServiceSharing; |
21 | 21 | import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; |
22 | 22 | import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; |
| 23 | +import org.apache.flink.table.api.DataTypes; |
| 24 | +import org.apache.flink.table.data.RowData; |
| 25 | +import org.apache.flink.table.data.conversion.RowRowConverter; |
| 26 | +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; |
| 27 | +import org.apache.flink.table.types.DataType; |
| 28 | +import org.apache.flink.table.types.logical.LogicalType; |
| 29 | +import org.apache.flink.table.types.logical.RowType; |
| 30 | +import org.apache.flink.table.types.utils.TypeConversions; |
23 | 31 | import org.apache.flink.test.util.MiniClusterWithClientResource; |
| 32 | +import org.apache.flink.util.CloseableIterator; |
24 | 33 |
|
25 | 34 | import com.ververica.cdc.connectors.base.experimental.MySqlSourceBuilder; |
| 35 | +import com.ververica.cdc.connectors.base.experimental.utils.MySqlConnectionUtils; |
26 | 36 | import com.ververica.cdc.connectors.base.source.JdbcIncrementalSource; |
27 | 37 | import com.ververica.cdc.connectors.base.testutils.MySqlContainer; |
28 | 38 | import com.ververica.cdc.connectors.base.testutils.MySqlVersion; |
29 | 39 | import com.ververica.cdc.connectors.base.testutils.UniqueDatabase; |
30 | 40 | import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema; |
| 41 | +import com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema; |
| 42 | +import io.debezium.connector.mysql.MySqlConnection; |
| 43 | +import io.debezium.jdbc.JdbcConnection; |
31 | 44 | import org.junit.BeforeClass; |
32 | 45 | import org.junit.Ignore; |
33 | 46 | import org.junit.Rule; |
|
37 | 50 | import org.testcontainers.containers.output.Slf4jLogConsumer; |
38 | 51 | import org.testcontainers.lifecycle.Startables; |
39 | 52 |
|
| 53 | +import java.sql.SQLException; |
| 54 | +import java.time.ZoneId; |
| 55 | +import java.util.ArrayList; |
| 56 | +import java.util.Arrays; |
| 57 | +import java.util.HashMap; |
| 58 | +import java.util.List; |
| 59 | +import java.util.Map; |
| 60 | +import java.util.stream.Collectors; |
40 | 61 | import java.util.stream.Stream; |
41 | 62 |
|
| 63 | +import static org.junit.Assert.assertArrayEquals; |
| 64 | +import static org.junit.Assert.assertEquals; |
| 65 | +import static org.junit.Assert.assertTrue; |
| 66 | + |
42 | 67 | /** Example Tests for {@link JdbcIncrementalSource}. */ |
43 | 68 | public class MySqlChangeEventSourceExampleTest { |
44 | 69 |
|
@@ -70,7 +95,7 @@ public static void startContainers() { |
70 | 95 |
|
71 | 96 | @Test |
72 | 97 | @Ignore("Test ignored because it won't stop and is used for manual test") |
73 | | - public void testConsumingAllEvents() throws Exception { |
| 98 | + public void testConsumingScanEvents() throws Exception { |
74 | 99 | inventoryDatabase.createAndInitialize(); |
75 | 100 | JdbcIncrementalSource<String> mySqlChangeEventSource = |
76 | 101 | new MySqlSourceBuilder() |
@@ -100,6 +125,154 @@ public void testConsumingAllEvents() throws Exception { |
100 | 125 | env.execute("Print MySQL Snapshot + Binlog"); |
101 | 126 | } |
102 | 127 |
|
| 128 | + @Test |
| 129 | + @Ignore("Test ignored because it won't stop and is used for manual test") |
| 130 | + public void testConsumingAllEvents() throws Exception { |
| 131 | + final DataType dataType = |
| 132 | + DataTypes.ROW( |
| 133 | + DataTypes.FIELD("id", DataTypes.BIGINT()), |
| 134 | + DataTypes.FIELD("name", DataTypes.STRING()), |
| 135 | + DataTypes.FIELD("description", DataTypes.STRING()), |
| 136 | + DataTypes.FIELD("weight", DataTypes.FLOAT())); |
| 137 | + |
| 138 | + inventoryDatabase.createAndInitialize(); |
| 139 | + final String tableId = inventoryDatabase.getDatabaseName() + ".products"; |
| 140 | + JdbcIncrementalSource<RowData> mySqlChangeEventSource = |
| 141 | + new MySqlSourceBuilder() |
| 142 | + .hostname(MYSQL_CONTAINER.getHost()) |
| 143 | + .port(MYSQL_CONTAINER.getDatabasePort()) |
| 144 | + .databaseList(inventoryDatabase.getDatabaseName()) |
| 145 | + .tableList(tableId) |
| 146 | + .username(inventoryDatabase.getUsername()) |
| 147 | + .password(inventoryDatabase.getPassword()) |
| 148 | + .serverId("5401-5404") |
| 149 | + .deserializer(buildRowDataDebeziumDeserializeSchema(dataType)) |
| 150 | + .includeSchemaChanges(true) // output the schema changes as well |
| 151 | + .splitSize(2) |
| 152 | + .build(); |
| 153 | + |
| 154 | + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); |
| 155 | + |
| 156 | + // enable checkpoint |
| 157 | + env.enableCheckpointing(3000); |
| 158 | + // set the source parallelism to 4 |
| 159 | + CloseableIterator<RowData> iterator = |
| 160 | + env.fromSource( |
| 161 | + mySqlChangeEventSource, |
| 162 | + WatermarkStrategy.noWatermarks(), |
| 163 | + "MySqlParallelSource") |
| 164 | + .setParallelism(4) |
| 165 | + .executeAndCollect(); // collect record |
| 166 | + |
| 167 | + String[] snapshotExpectedRecords = |
| 168 | + new String[] { |
| 169 | + "+I[101, scooter, Small 2-wheel scooter, 3.14]", |
| 170 | + "+I[102, car battery, 12V car battery, 8.1]", |
| 171 | + "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.8]", |
| 172 | + "+I[104, hammer, 12oz carpenter's hammer, 0.75]", |
| 173 | + "+I[105, hammer, 14oz carpenter's hammer, 0.875]", |
| 174 | + "+I[106, hammer, 16oz carpenter's hammer, 1.0]", |
| 175 | + "+I[107, rocks, box of assorted rocks, 5.3]", |
| 176 | + "+I[108, jacket, water resistent black wind breaker, 0.1]", |
| 177 | + "+I[109, spare tire, 24 inch spare tire, 22.2]" |
| 178 | + }; |
| 179 | + |
| 180 | + // step-1: consume snapshot data |
| 181 | + List<RowData> snapshotRowDataList = new ArrayList(); |
| 182 | + for (int i = 0; i < snapshotExpectedRecords.length && iterator.hasNext(); i++) { |
| 183 | + snapshotRowDataList.add(iterator.next()); |
| 184 | + } |
| 185 | + |
| 186 | + List<String> snapshotActualRecords = formatResult(snapshotRowDataList, dataType); |
| 187 | + assertEqualsInAnyOrder(Arrays.asList(snapshotExpectedRecords), snapshotActualRecords); |
| 188 | + |
| 189 | + // step-2: make 6 change events in one MySQL transaction |
| 190 | + makeBinlogEvents(getConnection(), tableId); |
| 191 | + |
| 192 | + String[] binlogExpectedRecords = |
| 193 | + new String[] { |
| 194 | + "-U[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.8]", |
| 195 | + "+U[103, cart, 12-pack of drill bits with sizes ranging from #40 to #3, 0.8]", |
| 196 | + "+I[110, spare tire, 28 inch spare tire, 26.2]", |
| 197 | + "-D[110, spare tire, 28 inch spare tire, 26.2]", |
| 198 | + "-U[103, cart, 12-pack of drill bits with sizes ranging from #40 to #3, 0.8]", |
| 199 | + "+U[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.8]" |
| 200 | + }; |
| 201 | + |
| 202 | + // step-3: consume binlog change events |
| 203 | + List<RowData> binlogRowDataList = new ArrayList(); |
| 204 | + for (int i = 0; i < binlogExpectedRecords.length && iterator.hasNext(); i++) { |
| 205 | + binlogRowDataList.add(iterator.next()); |
| 206 | + } |
| 207 | + List<String> binlogActualRecords = formatResult(binlogRowDataList, dataType); |
| 208 | + assertEqualsInAnyOrder(Arrays.asList(binlogExpectedRecords), binlogActualRecords); |
| 209 | + |
| 210 | + // stop the worker |
| 211 | + iterator.close(); |
| 212 | + } |
| 213 | + |
| 214 | + private RowDataDebeziumDeserializeSchema buildRowDataDebeziumDeserializeSchema( |
| 215 | + DataType dataType) { |
| 216 | + LogicalType logicalType = TypeConversions.fromDataToLogicalType(dataType); |
| 217 | + InternalTypeInfo<RowData> typeInfo = InternalTypeInfo.of(logicalType); |
| 218 | + return RowDataDebeziumDeserializeSchema.newBuilder() |
| 219 | + .setPhysicalRowType((RowType) dataType.getLogicalType()) |
| 220 | + .setResultTypeInfo(typeInfo) |
| 221 | + .build(); |
| 222 | + } |
| 223 | + |
| 224 | + private List<String> formatResult(List<RowData> records, DataType dataType) { |
| 225 | + RowRowConverter rowRowConverter = RowRowConverter.create(dataType); |
| 226 | + rowRowConverter.open(Thread.currentThread().getContextClassLoader()); |
| 227 | + return records.stream() |
| 228 | + .map(rowRowConverter::toExternal) |
| 229 | + .map(Object::toString) |
| 230 | + .collect(Collectors.toList()); |
| 231 | + } |
| 232 | + |
| 233 | + private MySqlConnection getConnection() { |
| 234 | + Map<String, String> properties = new HashMap<>(); |
| 235 | + properties.put("database.hostname", MYSQL_CONTAINER.getHost()); |
| 236 | + properties.put("database.port", String.valueOf(MYSQL_CONTAINER.getDatabasePort())); |
| 237 | + properties.put("database.user", inventoryDatabase.getUsername()); |
| 238 | + properties.put("database.password", inventoryDatabase.getPassword()); |
| 239 | + properties.put("database.serverTimezone", ZoneId.of("UTC").toString()); |
| 240 | + io.debezium.config.Configuration configuration = |
| 241 | + io.debezium.config.Configuration.from(properties); |
| 242 | + return MySqlConnectionUtils.createMySqlConnection(configuration); |
| 243 | + } |
| 244 | + |
| 245 | + private void makeBinlogEvents(JdbcConnection connection, String tableId) throws SQLException { |
| 246 | + try { |
| 247 | + connection.setAutoCommit(false); |
| 248 | + |
| 249 | + // make binlog events |
| 250 | + connection.execute( |
| 251 | + "UPDATE " + tableId + " SET name = 'cart' where id = 103", |
| 252 | + "INSERT INTO " |
| 253 | + + tableId |
| 254 | + + " VALUES(110,'spare tire','28 inch spare tire','26.2')", |
| 255 | + "DELETE FROM " + tableId + " where id = 110", |
| 256 | + "UPDATE " + tableId + " SET name = '12-pack drill bits' where id = 103"); |
| 257 | + connection.commit(); |
| 258 | + } finally { |
| 259 | + connection.close(); |
| 260 | + } |
| 261 | + } |
| 262 | + |
| 263 | + public static void assertEqualsInAnyOrder(List<String> expected, List<String> actual) { |
| 264 | + assertTrue(expected != null && actual != null); |
| 265 | + assertEqualsInOrder( |
| 266 | + expected.stream().sorted().collect(Collectors.toList()), |
| 267 | + actual.stream().sorted().collect(Collectors.toList())); |
| 268 | + } |
| 269 | + |
| 270 | + public static void assertEqualsInOrder(List<String> expected, List<String> actual) { |
| 271 | + assertTrue(expected != null && actual != null); |
| 272 | + assertEquals(expected.size(), actual.size()); |
| 273 | + assertArrayEquals(expected.toArray(new String[0]), actual.toArray(new String[0])); |
| 274 | + } |
| 275 | + |
103 | 276 | private static MySqlContainer createMySqlContainer(MySqlVersion version) { |
104 | 277 | return (MySqlContainer) |
105 | 278 | new MySqlContainer(version) |
|
0 commit comments