From a5c3dc009a560fe4b87a0d122385fd3bb4209fba Mon Sep 17 00:00:00 2001
From: bvolpato <brunocvcunha@gmail.com>
Date: Wed, 28 Oct 2026 00:58:21 -0400
Subject: [PATCH] [JDBC] Make autosharding test inserts idempotent

---
 .../java/org/apache/beam/sdk/io/jdbc/JdbcIOIT.java    | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOIT.java b/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOIT.java
index 57074fdffe82..2ea437f4efa0 100644
--- a/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOIT.java
+++ b/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOIT.java
@@ -271,7 +271,11 @@ private PipelineResult runRead(String tableName) {
   @Test
   public void testWriteWithAutosharding() throws Exception {
     String firstTableName = DatabaseTestHelper.getTestTableName("JDBCIT_AUTOSHARD");
-    DatabaseTestHelper.createTable(dataSource, firstTableName);
+    // JdbcIO may replay committed writes when a bundle is retried.
+    DatabaseTestHelper.createTable(
+        dataSource,
+        firstTableName,
+        Lists.newArrayList(KV.of("id", "INT PRIMARY KEY"), KV.of("name", "VARCHAR(500)")));
     try {
       PCollection<TestRow> dataCollection =
           pipelineWrite.apply(
@@ -282,7 +286,10 @@ public void testWriteWithAutosharding() throws Exception {
           .apply(
               JdbcIO.<TestRow>write()
                   .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(dataSource))
-                  .withStatement(String.format("insert into %s values(?, ?)", firstTableName))
+                  .withStatement(
+                      String.format(
+                          "insert into %s values(?, ?) on conflict (id) do nothing",
+                          firstTableName))
                   .withAutoSharding()
                   .withPreparedStatementSetter(new JdbcTestHelper.PrepareStatementFromTestRow()));