Skip to content

Commit

Permalink
use common abstraction for CDC via debezium for mysql (#4604)
Browse files Browse the repository at this point in the history 
* use new cdc abstraction for mysql

* undo wanted change

* pull in latest changes

* use renamed class + move constants to MySqlSource

* bring in latest changes from cdc abstraction

* format

* bring in latest changes

* pull in latest changes

* use common abstraction for CDC via debezium for postgres (#4607)

* use cdc abstraction for postgres

* add files

* ready

* use renamed class + move constants to PostgresSource

* bring in the latest changes

* bring in latest changes

* pull in latest changes
  • Loading branch information
@subodh1810
subodh1810 authored Jul 12, 2021
1 parent e542b89 commit fbc318f
Show file tree
Hide file tree
Showing 39 changed files with 862 additions and 3,164 deletions.
39 changes: 21 additions & 18 deletions .../bases/debezium/src/testFixtures/java/io/airbyte/integrations/debezium/CdcSourceTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,13 +71,13 @@ public abstract class CdcSourceTest {

private static final Logger LOGGER = LoggerFactory.getLogger(CdcSourceTest.class);

private static final String MODELS_SCHEMA = "models_schema";
private static final String MODELS_STREAM_NAME = "models";
protected static final String MODELS_SCHEMA = "models_schema";
protected static final String MODELS_STREAM_NAME = "models";
private static final Set<String> STREAM_NAMES = Sets
.newHashSet(MODELS_STREAM_NAME);
private static final String COL_ID = "id";
private static final String COL_MAKE_ID = "make_id";
private static final String COL_MODEL = "model";
protected static final String COL_ID = "id";
protected static final String COL_MAKE_ID = "make_id";
protected static final String COL_MODEL = "model";
protected static final String DB_NAME = MODELS_SCHEMA;

private static final AirbyteCatalog CATALOG = new AirbyteCatalog().withStreams(List.of(
Expand All @@ -89,7 +89,7 @@ public abstract class CdcSourceTest {
Field.of(COL_MODEL, JsonSchemaPrimitive.STRING))
.withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))
.withSourceDefinedPrimaryKey(List.of(List.of(COL_ID)))));
private static final ConfiguredAirbyteCatalog CONFIGURED_CATALOG = CatalogHelpers
protected static final ConfiguredAirbyteCatalog CONFIGURED_CATALOG = CatalogHelpers
.toDefaultConfiguredCatalog(CATALOG);

// set all streams to incremental.
Expand All @@ -105,7 +105,7 @@ public abstract class CdcSourceTest {
Jsons.jsonNode(ImmutableMap.of(COL_ID, 15, COL_MAKE_ID, 2, COL_MODEL, "A 220")),
Jsons.jsonNode(ImmutableMap.of(COL_ID, 16, COL_MAKE_ID, 2, COL_MODEL, "E 350")));

protected void setup() {
protected void setup() throws SQLException {
createAndPopulateTables();
}

Expand Down Expand Up @@ -155,8 +155,7 @@ private void createAndPopulateActualTable() {
*/
private void createAndPopulateRandomTable() {
createSchema(MODELS_SCHEMA + "_random");
createTable(MODELS_SCHEMA + "_random",
MODELS_STREAM_NAME + "_random",
createTable(MODELS_SCHEMA + "_random", MODELS_STREAM_NAME + "_random",
String.format("%s INTEGER, %s INTEGER, %s VARCHAR(200), PRIMARY KEY (%s)", COL_ID + "_random",
COL_MAKE_ID + "_random",
COL_MODEL + "_random", COL_ID + "_random"));
Expand Down Expand Up @@ -284,13 +283,13 @@ private void assertExpectedRecords(Set<JsonNode> expectedRecords,
@Test
@DisplayName("On the first sync, produce returns records that exist in the database.")
void testExistingData() throws Exception {
CdcTargetPosition targetPosition = cdcLatestTargetPosition();
final AutoCloseableIterator<AirbyteMessage> read = getSource().read(getConfig(), CONFIGURED_CATALOG, null);
final List<AirbyteMessage> actualRecords = AutoCloseableIterators.toListAndClose(read);

final Set<AirbyteRecordMessage> recordMessages = extractRecordMessages(actualRecords);
final List<AirbyteStateMessage> stateMessages = extractStateMessages(actualRecords);

CdcTargetPosition targetPosition = cdcLatestTargetPosition();
assertNotNull(targetPosition);
recordMessages.forEach(record -> {
assertEquals(extractPosition(record.getData()), targetPosition);
Expand Down Expand Up @@ -559,6 +558,17 @@ void testCheck() throws Exception {

@Test
void testDiscover() throws Exception {
final AirbyteCatalog expectedCatalog = expectedCatalogForDiscover();
final AirbyteCatalog actualCatalog = getSource().discover(getConfig());

assertEquals(
expectedCatalog.getStreams().stream().sorted(Comparator.comparing(AirbyteStream::getName))
.collect(Collectors.toList()),
actualCatalog.getStreams().stream().sorted(Comparator.comparing(AirbyteStream::getName))
.collect(Collectors.toList()));
}

protected AirbyteCatalog expectedCatalogForDiscover() {
final AirbyteCatalog expectedCatalog = Jsons.clone(CATALOG);

createTable(MODELS_SCHEMA, MODELS_STREAM_NAME + "_2", String.format("%s INTEGER, %s INTEGER, %s VARCHAR(200)", COL_ID, COL_MAKE_ID, COL_MODEL));
Expand All @@ -580,14 +590,7 @@ void testDiscover() throws Exception {

streams.add(streamWithoutPK);
expectedCatalog.withStreams(streams);

final AirbyteCatalog actualCatalog = getSource().discover(getConfig());

assertEquals(
expectedCatalog.getStreams().stream().sorted(Comparator.comparing(AirbyteStream::getName))
.collect(Collectors.toList()),
actualCatalog.getStreams().stream().sorted(Comparator.comparing(AirbyteStream::getName))
.collect(Collectors.toList()));
return expectedCatalog;
}

protected abstract CdcTargetPosition cdcLatestTargetPosition();
Expand Down
5 changes: 0 additions & 5 deletions ...ors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,11 +62,6 @@ public abstract class AbstractJdbcSource extends AbstractRelationalDbSource<JDBC

private static final Logger LOGGER = LoggerFactory.getLogger(AbstractJdbcSource.class);

public static final String CDC_LSN = "_ab_cdc_lsn";

public static final String CDC_LOG_FILE = "_ab_cdc_log_file";
public static final String CDC_LOG_POS = "_ab_cdc_log_pos";

private static final String JDBC_COLUMN_DATABASE_NAME = "TABLE_CAT";
private static final String JDBC_COLUMN_SCHEMA_NAME = "TABLE_SCHEM";
private static final String JDBC_COLUMN_TABLE_NAME = "TABLE_NAME";
Expand Down
5 changes: 2 additions & 3 deletions airbyte-integrations/connectors/source-mysql/build.gradle
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,16 +11,15 @@ application {
dependencies {
implementation project(':airbyte-db')
implementation project(':airbyte-integrations:bases:base-java')
implementation project(':airbyte-integrations:bases:debezium')
implementation project(':airbyte-integrations:connectors:source-jdbc')
implementation project(':airbyte-protocol:models')
implementation project(':airbyte-integrations:connectors:source-relational-db')

implementation 'io.debezium:debezium-api:1.4.2.Final'
implementation 'io.debezium:debezium-connector-mysql:1.4.2.Final'
implementation 'io.debezium:debezium-embedded:1.4.2.Final'
implementation 'mysql:mysql-connector-java:8.0.22'
implementation 'org.apache.commons:commons-lang3:3.11'

testImplementation testFixtures(project(':airbyte-integrations:bases:debezium'))
testImplementation testFixtures(project(':airbyte-integrations:connectors:source-jdbc'))
testImplementation 'org.apache.commons:commons-lang3:3.11'
testImplementation 'org.testcontainers:mysql:1.15.1'
Expand Down
178 changes: 0 additions & 178 deletions ...sql/src/main/java/io/airbyte/integrations/source/mysql/AirbyteFileOffsetBackingStore.java
View file

This file was deleted.

Loading

0 comments on commit fbc318f

Please sign in to comment.