Add JDBC-based source coordination store plugin by lawofcycles · Pull Request #6757 · opensearch-project/data-prepper

lawofcycles · 2026-04-14T14:31:09Z

Description

Adds a new jdbc source coordination store plugin backed by a relational database.

The plugin uses standard SQL operations to implement the SourceCoordinationStore interface.

Capability	Implementation
Conditional create	`INSERT` with duplicate key check
Optimistic locking	`UPDATE ... WHERE version = ?`
Partition acquisition	Status-based query with priority ordering
TTL cleanup	Scheduled deletion of expired `COMPLETED` partitions

Configuration example

source_coordination:
 store:
   jdbc:
     url: "jdbc:postgresql://localhost:5432/dataprepper"
     username: "dp_user"
     password: "dp_pass"

Tested with PostgreSQL and MySQL. Integration tests run against PostgreSQL by default and can be configured for MySQL via system properties. Also verified end-to-end with OpenSearch Source and Iceberg Source pipelines on both databases, including TTL-based cleanup of completed partitions.

Issues Resolved

Resolves #6740

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: Sotaro Hikita <bering1814@gmail.com>

srikanthpadakanti · 2026-04-15T17:36:12Z

+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+


No unit tests for JdbcSourceCoordinationStore itself (only integration tests). The DynamoDB store has unit tests with mocked clients. Add unit tests that mock DataSource/Connection to test error handling paths, version mismatch behavior, and SQL exception translation without requiring a running database.

Added JdbcSourceCoordinationStoreTest.

srikanthpadakanti · 2026-04-15T17:38:48Z

+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;


Uses JUnit assertions (assertEquals, assertFalse, assertTrue). Project standard requires Hamcrest matchers (assertThat with equalTo, is(true), is(false)).

srikanthpadakanti · 2026-04-15T17:40:55Z

+
+        final Optional<SourcePartitionStoreItem> first = store.tryAcquireAvailablePartition(
+                SOURCE_ID, "node-1", Duration.ofSeconds(2));
+        assertTrue(first.isPresent());


Thread.sleep(3000) in test. Use org.awaitility:awaitility for polling-based assertions instead of fixed sleeps. The DynamoDB store tests already use awaitility as a test dependency.

srikanthpadakanti · 2026-04-15T17:41:37Z

+
+import java.time.Instant;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;


uses JUnit assertions instead of Hamcrest matchers

srikanthpadakanti · 2026-04-15T17:41:50Z

+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;


uses JUnit assertions instead of Hamcrest matchers

srikanthpadakanti · 2026-04-15T17:45:59Z

+    implementation 'com.zaxxer:HikariCP:5.1.0'
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    implementation 'javax.inject:javax.inject:1'
+    implementation 'org.postgresql:postgresql:42.7.7'


Any reason why both postgresql and mysql included as implementation dependencies?
As they should be runtimeOnly. No compile-time references to driver-specific classes exist in the source as they're loaded via DriverManager at runtime.

srikanthpadakanti · 2026-04-15T17:47:21Z

+    implementation 'org.postgresql:postgresql:42.7.7'
+    implementation 'com.mysql:mysql-connector-j:8.4.0'
+}
+


No JaCoCo configuration. New plugins should have jacocoTestCoverageVerification with minimum 1.0 ratio per project standards.
@dlvenable correct me if I am wrong.

Added jacocoTestCoverageVerification with minimum 0.99. The threshold varies across plugins (from 0.84 to 1.0), and the current instruction coverage is 99%. The only uncovered line is the HikariDataSource constructor in createDataSource(), which is overridden in tests to inject a mock.

srikanthpadakanti · 2026-04-15T17:49:41Z

+    @JsonCreator
+    public JdbcStoreSettings(
+            @JsonProperty("url") final String url,
+            @JsonProperty("username") final String username,


password is stored as plain String. Consider documenting that this field supports Data-Prepper's secret management integration, or note it as a known limitation for the initial version

The password field already supports Data Prepper's secret management through VariableExpander. Users can use ${{aws_secrets:secretId:secretKey}} syntax, which is automatically resolved during deserialization via @JsonProperty.
However, there is currently only AwsSecretsPluginConfigValueTranslator, so it would be worth adding another implementation such as HashiCorp Vault for cloud-agnostic use cases in the future.

srikanthpadakanti · 2026-04-15T17:57:41Z

+
+    private final JdbcStoreSettings settings;
+    private HikariDataSource dataSource;
+    private ScheduledExecutorService ttlExecutor;


HikariDataSource and ScheduledExecutorService are never shut down. Implement AutoCloseable with a close() that calls dataSource.close() and ttlExecutor.shutdownNow().

Addressed. Implemented AutoCloseable with close() that shuts down both ttlExecutor and dataSource.

srikanthpadakanti · 2026-04-15T18:00:30Z

+    }
+
+    @Override
+    public void initializeStore() {


initializeStore() is not guarded against concurrent calls. If called twice, it creates a second HikariDataSource without closing the first. Add an AtomicBoolean initialized guard or null check on dataSource.

Addressed. Added AtomicBoolean guard in initializeStore() to prevent double initialization.

srikanthpadakanti · 2026-04-15T18:03:36Z

+        }
+
+        if (settings.getTtl() != null) {
+            ttlExecutor = Executors.newSingleThreadScheduledExecutor(r -> {


The ScheduledExecutorService for TTL cleanup is never shut down. On pipeline stop this daemon thread keeps running. Store a reference and shut it down in the close() method.

…Store Implement AutoCloseable to shut down HikariDataSource and TTL executor on close. Add AtomicBoolean guard to prevent double initialization and connection pool leak when initializeStore is called more than once. Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

… awaitility Align test style with project standards. Add awaitility test dependency and docker-compose.yml with PostgreSQL and MySQL for integration tests. Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

…-compose Extract createDataSource() as package-private method to allow test subclass override, replacing reflection-based mock injection. Add JdbcSourceCoordinationStoreTest covering error handling, version mismatch, table creation, and TTL paths (34 test cases). Add jacocoTestCoverageVerification with minimum 0.90 ratio. Add docker-compose.yml with PostgreSQL 16 and MySQL 8.4 for integration tests. Replace FQCN usages with imports. Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Extract createDataSource() as package-private method to enable test subclass override without reflection. Add test cases for table creation, MySQL duplicate index, non-constraint SQL exceptions, assigned-expired acquisition, null closedCount, and priority override branches. Raise JaCoCo minimum from 0.90 to 0.99. Replace remaining FQCN usages with imports in source and test files. Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

lawofcycles · 2026-04-16T13:03:57Z

Thank you for the thorough review, @srikanthpadakanti. All comments have been addressed.

srikanthpadakanti · 2026-04-16T18:27:03Z

+    private void createTable() {
+        final String tableName = settings.getTableName();
+        final String createTableSql = "CREATE TABLE IF NOT EXISTS " + tableName + " ("
+                + "source_identifier VARCHAR(256) NOT NULL, "


source_identifier and source_partition_key are VARCHAR(256). S3 object keys can be up to 1024 characters and some source plugins may use them as partition keys. Consider increasing to VARCHAR(1024) or documenting the limitation.

srikanthpadakanti · 2026-04-16T18:29:15Z

Thank you for the thorough review, @srikanthpadakanti. All comments have been addressed.

Thanks for addressing them. Added one more comment. Everything else looks good to me.

Add JDBC-based source coordination store plugin

50e1412

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

lawofcycles requested review from KarstenSchnitter, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners April 14, 2026 14:31

srikanthpadakanti reviewed Apr 15, 2026

View reviewed changes

lawofcycles added 5 commits April 16, 2026 09:23

Change JDBC driver dependencies from implementation to runtimeOnly

4c40833

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Replace JUnit assertions with Hamcrest matchers and Thread.sleep with…

ed1a8e4

… awaitility Align test style with project standards. Add awaitility test dependency and docker-compose.yml with PostgreSQL and MySQL for integration tests. Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

srikanthpadakanti reviewed Apr 16, 2026

View reviewed changes


		import java.time.Instant;

		import static org.junit.jupiter.api.Assertions.assertEquals;

Conversation

lawofcycles commented Apr 14, 2026

Description

Issues Resolved

Check List

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lawofcycles Apr 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lawofcycles commented Apr 16, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srikanthpadakanti commented Apr 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

lawofcycles Apr 16, 2026 •

edited

Loading

srikanthpadakanti commented Apr 16, 2026 •

edited

Loading