Few updates to RDS source plugin

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<String, Map<String, String>> getColumnDataTypeMap(final SchemaManager schemaManager) {
-        TableFilterConfig tableFilterConfig = sourceConfig.getTables();
-        Set<String> tableNames = schemaManager.getTableNames(tableFilterConfig.getDatabase());
-        tableFilterConfig.applyTableFilter(tableNames);
-        LOG.info("These tables will be include in processing: {}", tableNames);
+        Set<String> tableNames = schemaManager.getTableNames(sourceConfig.getDatabase());
+        sourceConfig.applyTableFilter(tableNames);
+        LOG.info("These tables will be included in processing: {}", tableNames);
         return schemaManager.getColumnDataTypes(new ArrayList<>(tableNames));
     }
 }

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<String> tableNames) {
+        if (tableFilterConfig == null) {
+            return;
+        }
+
+        if (!tableFilterConfig.getInclude().isEmpty()) {
+            List<String> includeTableList = tableFilterConfig.getInclude().stream()
+                    .map(item -> getDatabase() + "." + item)
+                    .collect(Collectors.toList());
+            tableNames.retainAll(includeTableList);
+        }
+
+        if (!tableFilterConfig.getExclude().isEmpty()) {
+            List<String> excludeTableList = tableFilterConfig.getExclude().stream()
+                    .map(item -> getDatabase() + "." + item)
+                    .collect(Collectors.toList());
+            excludeTableList.forEach(tableNames::remove);
+        }
+    }
 }

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/configuration/TableFilterConfig.java

@@ -10,48 +10,20 @@
 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<String> include = Collections.emptyList();
 
     @JsonProperty("exclude")
     @Size(max = 1000, message = "Table filter list should not be more than 1000")
     private List<String> 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<String> tableNames) {
-        if (!getInclude().isEmpty()) {
-            List<String> includeTableList = getInclude().stream()
-                    .map(item -> getDatabase() + "." + item)
-                    .collect(Collectors.toList());
-            tableNames.retainAll(includeTableList);
-        }
-
-        if (!getExclude().isEmpty()) {
-            List<String> excludeTableList = getExclude().stream()
-                    .map(item -> getDatabase() + "." + item)
-                    .collect(Collectors.toList());
-            excludeTableList.forEach(tableNames::remove);
-        }
-    }
 }

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() {

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());
     }
 }

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<BinlogCoordinate> 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)));
                 }

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;
+    }
+}

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);
 

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<String> 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);