diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsService.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsService.java index ea6e7c34f9..3d305adce3 100644 --- a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsService.java +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsService.java @@ -14,7 +14,6 @@ import org.opensearch.dataprepper.model.plugin.PluginConfigObservable; import org.opensearch.dataprepper.model.record.Record; import org.opensearch.dataprepper.model.source.coordinator.enhanced.EnhancedSourceCoordinator; -import org.opensearch.dataprepper.plugins.source.rds.configuration.TableFilterConfig; import org.opensearch.dataprepper.plugins.source.rds.export.DataFileScheduler; import org.opensearch.dataprepper.plugins.source.rds.export.ExportScheduler; import org.opensearch.dataprepper.plugins.source.rds.export.ExportTaskManager; @@ -213,10 +212,9 @@ private DbTableMetadata getDbTableMetadata(final DbMetadata dbMetadata, final Sc } private Map> getColumnDataTypeMap(final SchemaManager schemaManager) { - TableFilterConfig tableFilterConfig = sourceConfig.getTables(); - Set tableNames = schemaManager.getTableNames(tableFilterConfig.getDatabase()); - tableFilterConfig.applyTableFilter(tableNames); - LOG.info("These tables will be include in processing: {}", tableNames); + Set tableNames = schemaManager.getTableNames(sourceConfig.getDatabase()); + sourceConfig.applyTableFilter(tableNames); + LOG.info("These tables will be included in processing: {}", tableNames); return schemaManager.getColumnDataTypes(new ArrayList<>(tableNames)); } } diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfig.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfig.java index d8ed7e5843..294542c24d 100644 --- a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfig.java +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfig.java @@ -10,6 +10,7 @@ import jakarta.validation.constraints.Max; import jakarta.validation.constraints.Min; import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotEmpty; import jakarta.validation.constraints.NotNull; import org.opensearch.dataprepper.plugins.source.rds.configuration.AwsAuthenticationConfig; import org.opensearch.dataprepper.plugins.source.rds.configuration.EngineType; @@ -19,6 +20,9 @@ import software.amazon.awssdk.regions.Region; import java.time.Duration; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; /** * Configuration for RDS Source @@ -43,6 +47,10 @@ public class RdsSourceConfig { @NotNull private EngineType engine; + @JsonProperty("database") + @NotEmpty + private String database; + @JsonProperty("tables") private TableFilterConfig tableFilterConfig; @@ -111,6 +119,10 @@ public boolean isAurora() { return engine.isAurora(); } + public String getDatabase() { + return database; + } + public TableFilterConfig getTables() { return tableFilterConfig; } @@ -190,4 +202,29 @@ public String getPassword() { return password; } } + + /** + * This method applies the table filter configuration to the given set of table names. + * + * @param tableNames The set of table names to be filtered + */ + public void applyTableFilter(Set tableNames) { + if (tableFilterConfig == null) { + return; + } + + if (!tableFilterConfig.getInclude().isEmpty()) { + List includeTableList = tableFilterConfig.getInclude().stream() + .map(item -> getDatabase() + "." + item) + .collect(Collectors.toList()); + tableNames.retainAll(includeTableList); + } + + if (!tableFilterConfig.getExclude().isEmpty()) { + List excludeTableList = tableFilterConfig.getExclude().stream() + .map(item -> getDatabase() + "." + item) + .collect(Collectors.toList()); + excludeTableList.forEach(tableNames::remove); + } + } } diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/configuration/TableFilterConfig.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/configuration/TableFilterConfig.java index 7ed7111d45..7693755de8 100644 --- a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/configuration/TableFilterConfig.java +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/configuration/TableFilterConfig.java @@ -10,22 +10,15 @@ package org.opensearch.dataprepper.plugins.source.rds.configuration; import com.fasterxml.jackson.annotation.JsonProperty; -import jakarta.validation.constraints.NotEmpty; import jakarta.validation.constraints.Size; import lombok.Getter; import java.util.Collections; import java.util.List; -import java.util.Set; -import java.util.stream.Collectors; @Getter public class TableFilterConfig { - @JsonProperty("database") - @NotEmpty - private String database; - @JsonProperty("include") @Size(max = 1000, message = "Table filter list should not be more than 1000") private List include = Collections.emptyList(); @@ -33,25 +26,4 @@ public class TableFilterConfig { @JsonProperty("exclude") @Size(max = 1000, message = "Table filter list should not be more than 1000") private List exclude = Collections.emptyList(); - - /** - * This method applies the table filter configuration to the given set of table names. - * - * @param tableNames The set of table names to be filtered - */ - public void applyTableFilter(Set tableNames) { - if (!getInclude().isEmpty()) { - List includeTableList = getInclude().stream() - .map(item -> getDatabase() + "." + item) - .collect(Collectors.toList()); - tableNames.retainAll(includeTableList); - } - - if (!getExclude().isEmpty()) { - List excludeTableList = getExclude().stream() - .map(item -> getDatabase() + "." + item) - .collect(Collectors.toList()); - excludeTableList.forEach(tableNames::remove); - } - } } diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/leader/LeaderScheduler.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/leader/LeaderScheduler.java index e45a7af091..d71356da84 100644 --- a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/leader/LeaderScheduler.java +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/leader/LeaderScheduler.java @@ -8,7 +8,6 @@ import org.opensearch.dataprepper.model.source.coordinator.enhanced.EnhancedSourceCoordinator; import org.opensearch.dataprepper.model.source.coordinator.enhanced.EnhancedSourcePartition; import org.opensearch.dataprepper.plugins.source.rds.RdsSourceConfig; -import org.opensearch.dataprepper.plugins.source.rds.configuration.EngineType; import org.opensearch.dataprepper.plugins.source.rds.coordination.partition.ExportPartition; import org.opensearch.dataprepper.plugins.source.rds.coordination.partition.GlobalState; import org.opensearch.dataprepper.plugins.source.rds.coordination.partition.LeaderPartition; @@ -116,21 +115,6 @@ public void run() { public void shutdown() { shutdownRequested = true; - - // Clean up publication and replication slot for Postgres - if (streamPartition != null) { - streamPartition.getProgressState().ifPresent(progressState -> { - if (EngineType.fromString(progressState.getEngineType()).isPostgres()) { - final PostgresStreamState postgresStreamState = progressState.getPostgresStreamState(); - final String publicationName = postgresStreamState.getPublicationName(); - final String replicationSlotName = postgresStreamState.getReplicationSlotName(); - LOG.info("Cleaned up logical replication slot {} and publication {}", - replicationSlotName, publicationName); - ((PostgresSchemaManager) schemaManager).deleteLogicalReplicationSlot( - publicationName, replicationSlotName); - } - }); - } } private void init() { diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/ConnectionManagerFactory.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/ConnectionManagerFactory.java index 2adc0d731f..d052f02a31 100644 --- a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/ConnectionManagerFactory.java +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/ConnectionManagerFactory.java @@ -38,6 +38,6 @@ public ConnectionManager getConnectionManager() { sourceConfig.getAuthenticationConfig().getUsername(), sourceConfig.getAuthenticationConfig().getPassword(), sourceConfig.isTlsEnabled(), - sourceConfig.getTables().getDatabase()); + sourceConfig.getDatabase()); } } diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManager.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManager.java index ed618d315e..2babcc8edc 100644 --- a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManager.java +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManager.java @@ -36,7 +36,9 @@ public class MySqlSchemaManager implements SchemaManager { static final String[] TABLE_TYPES = new String[]{"TABLE"}; static final String COLUMN_NAME = "COLUMN_NAME"; static final String TABLE_NAME = "TABLE_NAME"; + static final String MYSQL_VERSION_8_4 = "8.4"; static final String BINLOG_STATUS_QUERY = "SHOW MASTER STATUS"; + static final String NEW_BINLOG_STATUS_QUERY = "SHOW BINARY LOG STATUS"; static final String BINLOG_FILE = "File"; static final String BINLOG_POSITION = "Position"; static final int NUM_OF_RETRIES = 3; @@ -153,8 +155,11 @@ public Optional getCurrentBinaryLogPosition() { int retry = 0; while (retry <= NUM_OF_RETRIES) { try (final Connection connection = connectionManager.getConnection()) { + final String mySqlVersion = connection.getMetaData().getDatabaseProductVersion(); final Statement statement = connection.createStatement(); - final ResultSet rs = statement.executeQuery(BINLOG_STATUS_QUERY); + final ResultSet rs = VersionUtil.compareVersions(mySqlVersion, MYSQL_VERSION_8_4) >= 0 ? + statement.executeQuery(NEW_BINLOG_STATUS_QUERY) : + statement.executeQuery(BINLOG_STATUS_QUERY); if (rs.next()) { return Optional.of(new BinlogCoordinate(rs.getString(BINLOG_FILE), rs.getLong(BINLOG_POSITION))); } diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/VersionUtil.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/VersionUtil.java new file mode 100644 index 0000000000..79356c8173 --- /dev/null +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/VersionUtil.java @@ -0,0 +1,35 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.source.rds.schema; + +public class VersionUtil { + /* Compares two version strings. + * Returns -1 if version1 is less than version2, 0 if they are equal, and 1 if version1 is greater than version2. + */ + public static int compareVersions(String version1, String version2) { + String[] v1Parts = version1.split("\\."); + String[] v2Parts = version2.split("\\."); + + int maxLength = Math.max(v1Parts.length, v2Parts.length); + + for (int i = 0; i < maxLength; i++) { + int v1Part = i < v1Parts.length ? Integer.parseInt(v1Parts[i]) : 0; + int v2Part = i < v2Parts.length ? Integer.parseInt(v2Parts[i]) : 0; + + if (v1Part < v2Part) { + return -1; + } + if (v1Part > v2Part) { + return 1; + } + } + return 0; + } +} diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/LogicalReplicationEventProcessor.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/LogicalReplicationEventProcessor.java index 8f00125904..0ea43e2dd7 100644 --- a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/LogicalReplicationEventProcessor.java +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/LogicalReplicationEventProcessor.java @@ -226,7 +226,7 @@ void processBeginMessage(ByteBuffer msg) { void processRelationMessage(ByteBuffer msg) { final long tableId = msg.getInt(); - String databaseName = sourceConfig.getTables().getDatabase(); + String databaseName = sourceConfig.getDatabase(); String schemaName = getNullTerminatedString(msg); String tableName = getNullTerminatedString(msg); diff --git a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/RdsServiceTest.java b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/RdsServiceTest.java index cd246bd359..8a59968594 100644 --- a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/RdsServiceTest.java +++ b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/RdsServiceTest.java @@ -23,7 +23,6 @@ import org.opensearch.dataprepper.model.record.Record; import org.opensearch.dataprepper.model.source.coordinator.enhanced.EnhancedSourceCoordinator; import org.opensearch.dataprepper.plugins.source.rds.configuration.EngineType; -import org.opensearch.dataprepper.plugins.source.rds.configuration.TableFilterConfig; import org.opensearch.dataprepper.plugins.source.rds.configuration.TlsConfig; import org.opensearch.dataprepper.plugins.source.rds.export.DataFileScheduler; import org.opensearch.dataprepper.plugins.source.rds.export.ExportScheduler; @@ -196,13 +195,11 @@ private void prepareMocks() { .build()) .build()) .build(); - final TableFilterConfig tableFilterConfig = mock(TableFilterConfig.class); final String databaseName = UUID.randomUUID().toString(); final Set tableNames = Set.of("database1.table1", "database2.table2"); when(sourceConfig.getDbIdentifier()).thenReturn(dbIdentifier); - when(sourceConfig.getTables()).thenReturn(tableFilterConfig); - when(tableFilterConfig.getDatabase()).thenReturn(databaseName); + when(sourceConfig.getDatabase()).thenReturn(databaseName); when(schemaManager.getTableNames(databaseName)).thenReturn(tableNames); when(schemaManager.getColumnDataTypes(new ArrayList<>(tableNames))).thenReturn(mock(Map.class)); when(rdsClient.describeDBInstances(any(DescribeDbInstancesRequest.class))).thenReturn(describeDbInstancesResponse); diff --git a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfigTest.java b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfigTest.java index 678cab305a..b1009a0fd1 100644 --- a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfigTest.java +++ b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/RdsSourceConfigTest.java @@ -5,36 +5,49 @@ package org.opensearch.dataprepper.plugins.source.rds; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.ArgumentsProvider; +import org.junit.jupiter.params.provider.ArgumentsSource; import org.opensearch.dataprepper.plugins.source.rds.configuration.ExportConfig; +import org.opensearch.dataprepper.plugins.source.rds.configuration.TableFilterConfig; +import java.util.HashSet; +import java.util.List; +import java.util.Set; import java.util.UUID; +import java.util.stream.Stream; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; import static org.opensearch.dataprepper.test.helper.ReflectivelySetField.setField; class RdsSourceConfigTest { + + private RdsSourceConfig objectUnderTest; + + @BeforeEach + void setup() { + objectUnderTest = createObjectUnderTest(); + } @Test void test_default_RdsSourceConfig_returns_default_values() { - final RdsSourceConfig objectUnderTest = createObjectUnderTest(); - assertThat(objectUnderTest.isAcknowledgmentsEnabled(), equalTo(true)); assertThat(objectUnderTest.isDisableS3ReadForLeader(), equalTo(false)); } @Test void test_when_export_is_not_configured_then_isExportEnabled_returns_false() { - final RdsSourceConfig objectUnderTest = createObjectUnderTest(); - assertThat(objectUnderTest.isExportEnabled(), equalTo(false)); } @Test void test_when_export_is_configured_then_isExportEnabled_returns_true() throws NoSuchFieldException, IllegalAccessException { - final RdsSourceConfig objectUnderTest = createObjectUnderTest(); - ExportConfig exportConfig = new ExportConfig(); setField(ExportConfig.class, exportConfig, "kmsKeyId", UUID.randomUUID().toString()); setField(RdsSourceConfig.class, objectUnderTest, "exportConfig", exportConfig); @@ -42,7 +55,81 @@ void test_when_export_is_configured_then_isExportEnabled_returns_true() throws N assertThat(objectUnderTest.isExportEnabled(), equalTo(true)); } + @ParameterizedTest + @ArgumentsSource(MySqlTableFilterTestArgumentsProvider.class) + void test_filter_mysql(List includeTables, List excludeTables, Set expectedTables) throws NoSuchFieldException, IllegalAccessException { + final Set allTables = new HashSet<>(Set.of("database1.table1", "database1.table2", "database1.table3")); + TableFilterConfig tableFilterConfig = new TableFilterConfig(); + setField(RdsSourceConfig.class, objectUnderTest, "database", "database1"); + setField(RdsSourceConfig.class, objectUnderTest, "tableFilterConfig", tableFilterConfig); + setField(TableFilterConfig.class, tableFilterConfig, "include", includeTables); + setField(TableFilterConfig.class, tableFilterConfig, "exclude", excludeTables); + + objectUnderTest.applyTableFilter(allTables); + + assertThat(expectedTables, is(allTables)); + } + + @ParameterizedTest + @ArgumentsSource(PostgresTableFilterTestArgumentsProvider.class) + void test_filter_postgres(List includeTables, List excludeTables, Set expectedTables) throws NoSuchFieldException, IllegalAccessException { + final Set allTables = new HashSet<>(Set.of("database1.schema1.table1", "database1.schema1.table2", "database1.schema1.table3")); + TableFilterConfig tableFilterConfig = new TableFilterConfig(); + setField(RdsSourceConfig.class, objectUnderTest, "database", "database1"); + setField(RdsSourceConfig.class, objectUnderTest, "tableFilterConfig", tableFilterConfig); + setField(TableFilterConfig.class, tableFilterConfig, "include", includeTables); + setField(TableFilterConfig.class, tableFilterConfig, "exclude", excludeTables); + + objectUnderTest.applyTableFilter(allTables); + + assertThat(expectedTables, is(allTables)); + } + + static class MySqlTableFilterTestArgumentsProvider implements ArgumentsProvider { + @Override + public Stream provideArguments(ExtensionContext extensionContext) { + return Stream.of( + // Include only + Arguments.of(List.of("table1", "table2"), List.of(), Set.of("database1.table1", "database1.table2")), + Arguments.of(List.of("table1", "table4"), List.of(), Set.of("database1.table1")), + + // Exclude only + Arguments.of(List.of(), List.of("table1", "table2"), Set.of("database1.table3")), + Arguments.of(List.of(), List.of("table1", "table4"), Set.of("database1.table2", "database1.table3")), + + // Both include and exclude + Arguments.of(List.of("table1", "table2"), List.of("table1", "table2"), Set.of()), + Arguments.of(List.of("table1", "table2"), List.of("table2", "table3"), Set.of("database1.table1")), + + // No include or exclude + Arguments.of(List.of(), List.of(), Set.of("database1.table1", "database1.table2", "database1.table3")) + ); + } + } + + static class PostgresTableFilterTestArgumentsProvider implements ArgumentsProvider { + @Override + public Stream provideArguments(ExtensionContext extensionContext) { + return Stream.of( + // Include only + Arguments.of(List.of("schema1.table1", "schema1.table2"), List.of(), Set.of("database1.schema1.table1", "database1.schema1.table2")), + Arguments.of(List.of("schema1.table1", "schema1.table4"), List.of(), Set.of("database1.schema1.table1")), + + // Exclude only + Arguments.of(List.of(), List.of("schema1.table1", "schema1.table2"), Set.of("database1.schema1.table3")), + Arguments.of(List.of(), List.of("schema1.table1", "schema1.table4"), Set.of("database1.schema1.table2", "database1.schema1.table3")), + + // Both include and exclude + Arguments.of(List.of("schema1.table1", "schema1.table2"), List.of("schema1.table1", "schema1.table2"), Set.of()), + Arguments.of(List.of("schema1.table1", "schema1.table2"), List.of("schema1.table2", "schema1.table3", "schema2.table2"), Set.of("database1.schema1.table1")), + + // No include or exclude + Arguments.of(List.of(), List.of(), Set.of("database1.schema1.table1", "database1.schema1.table2", "database1.schema1.table3")) + ); + } + } + private RdsSourceConfig createObjectUnderTest() { return new RdsSourceConfig(); } -} \ No newline at end of file +} diff --git a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/configuration/TableFilterConfigTest.java b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/configuration/TableFilterConfigTest.java deleted file mode 100644 index 18ad27c308..0000000000 --- a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/configuration/TableFilterConfigTest.java +++ /dev/null @@ -1,106 +0,0 @@ -/* - * Copyright OpenSearch Contributors - * SPDX-License-Identifier: Apache-2.0 - * - * The OpenSearch Contributors require contributions made to - * this file be licensed under the Apache-2.0 license or a - * compatible open source license. - */ - -package org.opensearch.dataprepper.plugins.source.rds.configuration; - -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.extension.ExtensionContext; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.Arguments; -import org.junit.jupiter.params.provider.ArgumentsProvider; -import org.junit.jupiter.params.provider.ArgumentsSource; -import org.opensearch.dataprepper.test.helper.ReflectivelySetField; - -import java.util.HashSet; -import java.util.List; -import java.util.Set; -import java.util.stream.Stream; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.is; - - -class TableFilterConfigTest { - private TableFilterConfig tableFilterConfig; - - @BeforeEach - void setUp() { - tableFilterConfig = new TableFilterConfig(); - } - - @ParameterizedTest - @ArgumentsSource(MySqlTableFilterTestArgumentsProvider.class) - void test_filter_mysql(List includeTables, List excludeTables, Set expectedTables) throws NoSuchFieldException, IllegalAccessException { - final Set allTables = new HashSet<>(Set.of("database1.table1", "database1.table2", "database1.table3")); - ReflectivelySetField.setField(tableFilterConfig.getClass(), tableFilterConfig, "database", "database1"); - ReflectivelySetField.setField(tableFilterConfig.getClass(), tableFilterConfig, "include", includeTables); - ReflectivelySetField.setField(tableFilterConfig.getClass(), tableFilterConfig, "exclude", excludeTables); - - tableFilterConfig.applyTableFilter(allTables); - - assertThat(expectedTables, is(allTables)); - } - - @ParameterizedTest - @ArgumentsSource(PostgresTableFilterTestArgumentsProvider.class) - void test_filter_postgres(List includeTables, List excludeTables, Set expectedTables) throws NoSuchFieldException, IllegalAccessException { - final Set allTables = new HashSet<>(Set.of("database1.schema1.table1", "database1.schema1.table2", "database1.schema1.table3")); - ReflectivelySetField.setField(tableFilterConfig.getClass(), tableFilterConfig, "database", "database1"); - ReflectivelySetField.setField(tableFilterConfig.getClass(), tableFilterConfig, "include", includeTables); - ReflectivelySetField.setField(tableFilterConfig.getClass(), tableFilterConfig, "exclude", excludeTables); - - tableFilterConfig.applyTableFilter(allTables); - - assertThat(expectedTables, is(allTables)); - } - - static class MySqlTableFilterTestArgumentsProvider implements ArgumentsProvider { - @Override - public Stream provideArguments(ExtensionContext extensionContext) { - return Stream.of( - // Include only - Arguments.of(List.of("table1", "table2"), List.of(), Set.of("database1.table1", "database1.table2")), - Arguments.of(List.of("table1", "table4"), List.of(), Set.of("database1.table1")), - - // Exclude only - Arguments.of(List.of(), List.of("table1", "table2"), Set.of("database1.table3")), - Arguments.of(List.of(), List.of("table1", "table4"), Set.of("database1.table2", "database1.table3")), - - // Both include and exclude - Arguments.of(List.of("table1", "table2"), List.of("table1", "table2"), Set.of()), - Arguments.of(List.of("table1", "table2"), List.of("table2", "table3"), Set.of("database1.table1")), - - // No include or exclude - Arguments.of(List.of(), List.of(), Set.of("database1.table1", "database1.table2", "database1.table3")) - ); - } - } - - static class PostgresTableFilterTestArgumentsProvider implements ArgumentsProvider { - @Override - public Stream provideArguments(ExtensionContext extensionContext) { - return Stream.of( - // Include only - Arguments.of(List.of("schema1.table1", "schema1.table2"), List.of(), Set.of("database1.schema1.table1", "database1.schema1.table2")), - Arguments.of(List.of("schema1.table1", "schema1.table4"), List.of(), Set.of("database1.schema1.table1")), - - // Exclude only - Arguments.of(List.of(), List.of("schema1.table1", "schema1.table2"), Set.of("database1.schema1.table3")), - Arguments.of(List.of(), List.of("schema1.table1", "schema1.table4"), Set.of("database1.schema1.table2", "database1.schema1.table3")), - - // Both include and exclude - Arguments.of(List.of("schema1.table1", "schema1.table2"), List.of("schema1.table1", "schema1.table2"), Set.of()), - Arguments.of(List.of("schema1.table1", "schema1.table2"), List.of("schema1.table2", "schema1.table3", "schema2.table2"), Set.of("database1.schema1.table1")), - - // No include or exclude - Arguments.of(List.of(), List.of(), Set.of("database1.schema1.table1", "database1.schema1.table2", "database1.schema1.table3")) - ); - } - } -} \ No newline at end of file diff --git a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManagerTest.java b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManagerTest.java index fc27957ae7..382551dd59 100644 --- a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManagerTest.java +++ b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManagerTest.java @@ -135,9 +135,12 @@ void test_getTableNames_when_exception_then_throws() throws SQLException { @Test void test_getCurrentBinaryLogPosition_returns_binlog_coords() throws SQLException { final Statement statement = mock(Statement.class); + final DatabaseMetaData databaseMetaData = mock(DatabaseMetaData.class); final String binlogFile = UUID.randomUUID().toString(); final long binlogPosition = 123L; when(connectionManager.getConnection()).thenReturn(connection); + when(connection.getMetaData()).thenReturn(databaseMetaData); + when(databaseMetaData.getDatabaseProductVersion()).thenReturn("8.0"); when(connection.createStatement()).thenReturn(statement); when(statement.executeQuery(BINLOG_STATUS_QUERY)).thenReturn(resultSet); when(resultSet.next()).thenReturn(true); diff --git a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/schema/VersionUtilTest.java b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/schema/VersionUtilTest.java new file mode 100644 index 0000000000..1e89512650 --- /dev/null +++ b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/schema/VersionUtilTest.java @@ -0,0 +1,51 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.source.rds.schema; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.ArgumentsProvider; +import org.junit.jupiter.params.provider.ArgumentsSource; + +import java.util.stream.Stream; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class VersionUtilTest { + + @ParameterizedTest + @ArgumentsSource(VersionUtilTestCases.class) + void test_compareVersions(String version1, String version2, int expected) { + assertThat(VersionUtil.compareVersions(version1, version2), is(expected)); + } + + @Test + void test_invalidVersion_throws() { + assertThrows(NumberFormatException.class, () -> VersionUtil.compareVersions("1.0a", "1.0")); + } + + static class VersionUtilTestCases implements ArgumentsProvider { + @Override + public Stream provideArguments(ExtensionContext extensionContext) { + return Stream.of( + Arguments.of("1.0.0", "1.0.0", 0), + Arguments.of("1.0.0", "1.0.1", -1), + Arguments.of("1.0.1", "1.0.0", 1), + Arguments.of("1.0.0", "1.0", 0), + Arguments.of("1.0.1", "1.0", 1), + Arguments.of("1.0.1", "2.0", -1) + ); + } + } +}