Process resync partitions in RDS source #5171

oeyh · 2024-11-06T17:47:33Z

Description

Followup PR to #5168 to process the resync partitions.

Tested against some MySQL tables to verify that the source generated expected events.

Issues Resolved

Contributes to #4561

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: Hai Yan <[email protected]>

oeyh · 2024-11-12T22:04:54Z

Had to rebase to resolve merge conflicts.

...urce/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncScheduler.java

dinujoh · 2024-11-14T20:32:47Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

+        String[] keySplits = resyncPartition.getPartitionKey().split("\\|");
+        final String database = keySplits[0];
+        final String table = keySplits[1];
+        final long eventTimestampMillis = Long.parseLong(keySplits[2]);


Add validation on number of key splits

This can also be move to separate data structure and be created here.

final PartitionInfo partitionInfo = parsePartitionKey();

private static class PartitionInfo { final String database; final String table; final long eventTimestampMillis; PartitionInfo(String database, String table, long eventTimestampMillis) { this.database = database; this.table = table; this.eventTimestampMillis = eventTimestampMillis; } }

I made the change in ResyncPartition class and added unit tests.

dinujoh · 2024-11-14T20:39:11Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

+        String[] keySplits = resyncPartition.getPartitionKey().split("\\|");
+        final String database = keySplits[0];
+        final String table = keySplits[1];
+        final long eventTimestampMillis = Long.parseLong(keySplits[2]);


This can also be move to separate data structure and be created here.

final PartitionInfo partitionInfo = parsePartitionKey();

private static class PartitionInfo { final String database; final String table; final long eventTimestampMillis; PartitionInfo(String database, String table, long eventTimestampMillis) { this.database = database; this.table = table; this.eventTimestampMillis = eventTimestampMillis; } }

dinujoh · 2024-11-14T20:41:02Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

+        return new ResyncWorker(resyncPartition, sourceConfig, queryManager, buffer, recordConverter, acknowledgementSet);
+    }
+
+    public void run() {


This code can be modular. I will have some comments below to break down

dinujoh · 2024-11-14T20:42:20Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

+        final String table = keySplits[1];
+        final long eventTimestampMillis = Long.parseLong(keySplits[2]);
+
+        if (resyncPartition.getProgressState().isEmpty()) {


Move to separate method to validate and get progress state

final ResyncProgressState progressState = validateAndGetProgressState();

private ResyncProgressState validateAndGetProgressState() { return resyncPartition.getProgressState() .orElseThrow(() -> new IllegalStateException( "ResyncPartition " + resyncPartition.getPartitionKey() + " doesn't contain progress state.")); }

dinujoh · 2024-11-14T20:45:01Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

+            queryStatement = String.format("SELECT * FROM %s WHERE %s IS NULL", database + "." + table, foreignKeyName);
+        } else {
+            queryStatement = String.format("SELECT * FROM %s WHERE %s='%s'", database + "." + table, foreignKeyName, updatedValue);


Move the query to static constant. You can use SQL Prepared statement which is pre-compiled SQL statement.

Move this to separate method to build and execute query

List<Map<String, Object>> rows = executeQuery(); // or executeQuery(queryStatement)

dinujoh · 2024-11-14T20:45:55Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

+
+        BufferAccumulator<Record<Event>> bufferAccumulator = BufferAccumulator.create(buffer, DEFAULT_BUFFER_BATCH_SIZE, BUFFER_TIMEOUT);
+
+        for (Map<String, Object> row : rows) {


Move rest of the code to new method to process rows

processRows(rows, partitionInfo, progressState);

dinujoh · 2024-11-14T20:50:50Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/QueryManager.java

+            try (ResultSet resultSet = statement.executeQuery(query)) {
+                return convertResultSetToList(resultSet);
+            }
+        } catch (Exception e) {


We should consider adding retry if there is retryable error.

dinujoh · 2024-11-14T20:52:06Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

+        }
+        LOG.debug("Query statement: {}", queryStatement);
+
+        List<Map<String, Object>> rows = queryManager.selectRows(queryStatement);


These would need data type mapping.

Good catch. Will add.

Signed-off-by: Hai Yan <[email protected]>

dinujoh · 2024-11-15T22:43:39Z

...urce/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncScheduler.java

+    public void run() {
+        LOG.info("Start running Resync Scheduler");
+        ResyncPartition resyncPartition = null;
+        while (!shutdownRequested && !Thread.currentThread().isInterrupted()) {


We should add some metrics on when the resync occurs.

dinujoh · 2024-11-15T22:44:30Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

+        processRows(rows, database, table, primaryKeys, eventTimestampMillis);
+    }
+
+    private void processRows(List<Map<String, Object>> rows, String database, String table, List<String> primaryKeys, long eventTimestampMillis) {


Add metrics when records are processed with count

dinujoh · 2024-11-15T22:46:18Z

Thanks for the refactoring and changes. I had some comments on adding metrics to ReSync worker. You can address that as a separate issue.

oeyh mentioned this pull request Nov 6, 2024

Detect cascading actions in RDS source #5168

Merged

4 tasks

oeyh force-pushed the rds-resync-process branch from f620e9b to 07a1012 Compare November 11, 2024 18:09

oeyh marked this pull request as ready for review November 11, 2024 20:17

oeyh requested review from sb2k16, chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, KarstenSchnitter and dlvenable as code owners November 11, 2024 20:17

oeyh force-pushed the rds-resync-process branch from 07a1012 to de92832 Compare November 11, 2024 20:24

oeyh added 4 commits November 12, 2024 15:55

Process resync partition

fa5709f

Signed-off-by: Hai Yan <[email protected]>

Update tests

1fef189

Signed-off-by: Hai Yan <[email protected]>

Cleanup

cf1a566

Signed-off-by: Hai Yan <[email protected]>

Fix tests after rebase

9aea7db

Signed-off-by: Hai Yan <[email protected]>

oeyh force-pushed the rds-resync-process branch from de92832 to 9aea7db Compare November 12, 2024 22:03

chenqi0805 previously approved these changes Nov 13, 2024

View reviewed changes

dinujoh reviewed Nov 14, 2024

View reviewed changes

...urce/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncScheduler.java Outdated Show resolved Hide resolved

dinujoh reviewed Nov 14, 2024

View reviewed changes

Address comments

d40d45c

Signed-off-by: Hai Yan <[email protected]>

oeyh dismissed chenqi0805’s stale review via d40d45c November 15, 2024 20:34

dinujoh reviewed Nov 15, 2024

View reviewed changes

dinujoh approved these changes Nov 15, 2024

View reviewed changes

oeyh merged commit 2e32d3d into opensearch-project:main Nov 15, 2024
43 of 47 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Process resync partitions in RDS source #5171

Process resync partitions in RDS source #5171

oeyh commented Nov 6, 2024 •

edited

Loading

oeyh commented Nov 12, 2024

dinujoh Nov 14, 2024

dinujoh Nov 14, 2024

oeyh Nov 15, 2024

dinujoh Nov 14, 2024

dinujoh Nov 14, 2024

dinujoh Nov 14, 2024

oeyh Nov 15, 2024

dinujoh Nov 14, 2024

oeyh Nov 15, 2024

dinujoh Nov 14, 2024

oeyh Nov 15, 2024

dinujoh Nov 14, 2024

oeyh Nov 15, 2024

dinujoh Nov 14, 2024

oeyh Nov 14, 2024

oeyh Nov 15, 2024

dinujoh Nov 15, 2024

dinujoh Nov 15, 2024

dinujoh commented Nov 15, 2024


		BufferAccumulator<Record<Event>> bufferAccumulator = BufferAccumulator.create(buffer, DEFAULT_BUFFER_BATCH_SIZE, BUFFER_TIMEOUT);

		for (Map<String, Object> row : rows) {

Process resync partitions in RDS source #5171

Process resync partitions in RDS source #5171

Conversation

oeyh commented Nov 6, 2024 • edited Loading

Description

Issues Resolved

Check List

oeyh commented Nov 12, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dinujoh commented Nov 15, 2024

oeyh commented Nov 6, 2024 •

edited

Loading