-
Notifications
You must be signed in to change notification settings - Fork 3.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
MySQL Source: fixed unencrypted CDC connections (#18851)
* MySQL Source: fixed unencrypted CDC connections * updated changelog * bump version * auto-bump connector version Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>
- Loading branch information
1 parent
9231e3e
commit 8bb9701
Showing
7 changed files
with
204 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
198 changes: 198 additions & 0 deletions
198
...rations/io/airbyte/integration_tests/sources/CdcMySqlSslRequiredSourceAcceptanceTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,198 @@ | ||
| package io.airbyte.integrations.io.airbyte.integration_tests.sources; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.google.common.collect.ImmutableMap; | ||
| import com.google.common.collect.Iterables; | ||
| import com.google.common.collect.Lists; | ||
| import io.airbyte.commons.json.Jsons; | ||
| import io.airbyte.db.Database; | ||
| import io.airbyte.db.factory.DSLContextFactory; | ||
| import io.airbyte.db.factory.DatabaseDriver; | ||
| import io.airbyte.db.jdbc.JdbcUtils; | ||
| import io.airbyte.integrations.base.ssh.SshHelpers; | ||
| import io.airbyte.integrations.standardtest.source.SourceAcceptanceTest; | ||
| import io.airbyte.integrations.standardtest.source.TestDestinationEnv; | ||
| import io.airbyte.protocol.models.AirbyteMessage; | ||
| import io.airbyte.protocol.models.AirbyteRecordMessage; | ||
| import io.airbyte.protocol.models.AirbyteStateMessage; | ||
| import io.airbyte.protocol.models.CatalogHelpers; | ||
| import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; | ||
| import io.airbyte.protocol.models.ConfiguredAirbyteStream; | ||
| import io.airbyte.protocol.models.ConnectorSpecification; | ||
| import io.airbyte.protocol.models.DestinationSyncMode; | ||
| import io.airbyte.protocol.models.Field; | ||
| import io.airbyte.protocol.models.JsonSchemaType; | ||
| import io.airbyte.protocol.models.SyncMode; | ||
| import org.jooq.DSLContext; | ||
| import org.jooq.SQLDialect; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.testcontainers.containers.MySQLContainer; | ||
|
|
||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import static io.airbyte.integrations.io.airbyte.integration_tests.sources.utils.TestConstants.INITIAL_CDC_WAITING_SECONDS; | ||
| import static io.airbyte.protocol.models.SyncMode.INCREMENTAL; | ||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
| import static org.junit.jupiter.api.Assertions.assertFalse; | ||
|
|
||
| public class CdcMySqlSslRequiredSourceAcceptanceTest extends SourceAcceptanceTest { | ||
| private static final String STREAM_NAME = "id_and_name"; | ||
| private static final String STREAM_NAME2 = "starships"; | ||
| private MySQLContainer<?> container; | ||
| private JsonNode config; | ||
|
|
||
| @Override | ||
| protected String getImageName() { | ||
| return "airbyte/source-mysql:dev"; | ||
| } | ||
|
|
||
| @Override | ||
| protected ConnectorSpecification getSpec() throws Exception { | ||
| return SshHelpers.getSpecAndInjectSsh(); | ||
| } | ||
|
|
||
| @Override | ||
| protected JsonNode getConfig() { | ||
| return config; | ||
| } | ||
|
|
||
| @Override | ||
| protected ConfiguredAirbyteCatalog getConfiguredCatalog() { | ||
| return new ConfiguredAirbyteCatalog().withStreams(Lists.newArrayList( | ||
| new ConfiguredAirbyteStream() | ||
| .withSyncMode(INCREMENTAL) | ||
| .withDestinationSyncMode(DestinationSyncMode.APPEND) | ||
| .withStream(CatalogHelpers.createAirbyteStream( | ||
| String.format("%s", STREAM_NAME), | ||
| String.format("%s", config.get(JdbcUtils.DATABASE_KEY).asText()), | ||
| Field.of("id", JsonSchemaType.NUMBER), | ||
| Field.of("name", JsonSchemaType.STRING)) | ||
| .withSourceDefinedCursor(true) | ||
| .withSourceDefinedPrimaryKey(List.of(List.of("id"))) | ||
| .withSupportedSyncModes( | ||
| Lists.newArrayList(SyncMode.FULL_REFRESH, INCREMENTAL))), | ||
| new ConfiguredAirbyteStream() | ||
| .withSyncMode(INCREMENTAL) | ||
| .withDestinationSyncMode(DestinationSyncMode.APPEND) | ||
| .withStream(CatalogHelpers.createAirbyteStream( | ||
| String.format("%s", STREAM_NAME2), | ||
| String.format("%s", config.get(JdbcUtils.DATABASE_KEY).asText()), | ||
| Field.of("id", JsonSchemaType.NUMBER), | ||
| Field.of("name", JsonSchemaType.STRING)) | ||
| .withSourceDefinedCursor(true) | ||
| .withSourceDefinedPrimaryKey(List.of(List.of("id"))) | ||
| .withSupportedSyncModes( | ||
| Lists.newArrayList(SyncMode.FULL_REFRESH, INCREMENTAL))))); | ||
| } | ||
|
|
||
| @Override | ||
| protected JsonNode getState() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| protected void setupEnvironment(final TestDestinationEnv environment) throws Exception { | ||
| container = new MySQLContainer<>("mysql:8.0"); | ||
| container.start(); | ||
|
|
||
| final var sslMode = ImmutableMap.builder() | ||
| .put(JdbcUtils.MODE_KEY, "required") | ||
| .build(); | ||
| final JsonNode replicationMethod = Jsons.jsonNode(ImmutableMap.builder() | ||
| .put("method", "CDC") | ||
| .put("initial_waiting_seconds", INITIAL_CDC_WAITING_SECONDS) | ||
| .build()); | ||
|
|
||
| config = Jsons.jsonNode(ImmutableMap.builder() | ||
| .put(JdbcUtils.HOST_KEY, container.getHost()) | ||
| .put(JdbcUtils.PORT_KEY, container.getFirstMappedPort()) | ||
| .put(JdbcUtils.DATABASE_KEY, container.getDatabaseName()) | ||
| .put(JdbcUtils.USERNAME_KEY, container.getUsername()) | ||
| .put(JdbcUtils.PASSWORD_KEY, container.getPassword()) | ||
| .put(JdbcUtils.SSL_KEY, true) | ||
| .put(JdbcUtils.SSL_MODE_KEY, sslMode) | ||
| .put("replication_method", replicationMethod) | ||
| .put("is_test", true) | ||
| .build()); | ||
|
|
||
| revokeAllPermissions(); | ||
| grantCorrectPermissions(); | ||
| alterUserRequireSsl(); | ||
| createAndPopulateTables(); | ||
| } | ||
|
|
||
| private void alterUserRequireSsl() { | ||
| executeQuery("ALTER USER " + container.getUsername() + " REQUIRE SSL;"); | ||
| } | ||
|
|
||
| private void createAndPopulateTables() { | ||
| executeQuery("CREATE TABLE id_and_name(id INTEGER PRIMARY KEY, name VARCHAR(200));"); | ||
| executeQuery( | ||
| "INSERT INTO id_and_name (id, name) VALUES (1,'picard'), (2, 'crusher'), (3, 'vash');"); | ||
| executeQuery("CREATE TABLE starships(id INTEGER PRIMARY KEY, name VARCHAR(200));"); | ||
| executeQuery( | ||
| "INSERT INTO starships (id, name) VALUES (1,'enterprise-d'), (2, 'defiant'), (3, 'yamato');"); | ||
| } | ||
|
|
||
| private void revokeAllPermissions() { | ||
| executeQuery("REVOKE ALL PRIVILEGES, GRANT OPTION FROM " + container.getUsername() + "@'%';"); | ||
| } | ||
|
|
||
| private void grantCorrectPermissions() { | ||
| executeQuery( | ||
| "GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO " | ||
| + container.getUsername() + "@'%';"); | ||
| } | ||
|
|
||
| private void executeQuery(final String query) { | ||
| try (final DSLContext dslContext = DSLContextFactory.create( | ||
| "root", | ||
| "test", | ||
| DatabaseDriver.MYSQL.getDriverClassName(), | ||
| String.format(DatabaseDriver.MYSQL.getUrlFormatString(), | ||
| container.getHost(), | ||
| container.getFirstMappedPort(), | ||
| container.getDatabaseName()), | ||
| SQLDialect.MYSQL)) { | ||
| final Database database = new Database(dslContext); | ||
| database.query( | ||
| ctx -> ctx | ||
| .execute(query)); | ||
| } catch (final Exception e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| protected void tearDown(final TestDestinationEnv testEnv) { | ||
| container.close(); | ||
| } | ||
|
|
||
| @Test | ||
| public void testIncrementalSyncShouldNotFailIfBinlogIsDeleted() throws Exception { | ||
| final ConfiguredAirbyteCatalog configuredCatalog = withSourceDefinedCursors(getConfiguredCatalog()); | ||
| // only sync incremental streams | ||
| configuredCatalog.setStreams( | ||
| configuredCatalog.getStreams().stream().filter(s -> s.getSyncMode() == INCREMENTAL).collect(Collectors.toList())); | ||
|
|
||
| final List<AirbyteMessage> airbyteMessages = runRead(configuredCatalog, getState()); | ||
| final List<AirbyteRecordMessage> recordMessages = filterRecords(airbyteMessages); | ||
| final List<AirbyteStateMessage> stateMessages = airbyteMessages | ||
| .stream() | ||
| .filter(m -> m.getType() == AirbyteMessage.Type.STATE) | ||
| .map(AirbyteMessage::getState) | ||
| .collect(Collectors.toList()); | ||
| assertFalse(recordMessages.isEmpty(), "Expected the first incremental sync to produce records"); | ||
| assertFalse(stateMessages.isEmpty(), "Expected incremental sync to produce STATE messages"); | ||
|
|
||
| // when we run incremental sync again there should be no new records. Run a sync with the latest | ||
| // state message and assert no records were emitted. | ||
| final JsonNode latestState = Jsons.jsonNode(supportsPerStream() ? stateMessages : List.of(Iterables.getLast(stateMessages))); | ||
| // RESET MASTER removes all binary log files that are listed in the index file, | ||
| // leaving only a single, empty binary log file with a numeric suffix of .000001 | ||
| executeQuery("RESET MASTER;"); | ||
|
|
||
| assertEquals(6, filterRecords(runRead(configuredCatalog, latestState)).size()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters