msabiransari opened a new issue, #18743:
URL: https://github.com/apache/iceberg/issues/18743
### Apache Iceberg version
1.12.0 (latest release)
### Query engine
Other
### Please describe the bug 🐞
**Summary.** With `JdbcCatalog`, if the database connection drops *after*
PostgreSQL has applied the commit's `UPDATE ... SET metadata_location = ? WHERE
metadata_location = ?` but before the client reads the response,
`JdbcClientPool` (via `ClientPoolImpl.run`) reconnects and **replays the same
UPDATE**. The replay matches 0 rows because the compare-and-swap predicate no
longer holds, so `JdbcTableOperations.updateTable` throws
`CommitFailedException` for a commit that is already durable.
`CommitFailedException` is a `CleanableFailure`, so:
* with `commit.retry.num-retries=0`: `SnapshotProducer`/`BaseTransaction`
run `cleanAll()` and **delete the manifest list and manifests that the table's
current metadata now references**. The table is left unreadable
(`NotFoundException` on scan).
* with the default retries (4), single writer: the `Tasks` retry refreshes,
reuses the staged snapshot id, takes the rollback branch in
`SnapshotProducer.commit()` and returns normally, so it heals by accident.
* with the default retries and **a concurrent writer committing between the
lost acknowledgement and the retry**: both commits return normally, but `main`
is rewound to writer 1's snapshot and the other writer's committed data file is
no longer visible. Nothing throws.
Verified on Iceberg 1.12.0 against PostgreSQL 17.11 (JDK 21, PgJDBC 42.7.14,
`InMemoryFileIO`), five runs, deterministic. The relevant code is unchanged on
`main` as of 2742348d.
**Mechanism (1.12.0 sources).**
1. `ClientPoolImpl.run(action, retry)` re-runs the whole action after
`reconnect()` when `isConnectionException(exc)`
(`core/.../ClientPoolImpl.java:69-95`). `JdbcClientPool` passes `retryByDefault
= true` and treats SQLSTATE `08000/08003/08006/08007/40001` as connection
exceptions (`JdbcClientPool.java:50-51, 110-114`). A dropped PostgreSQL socket
surfaces as `08006`.
2. `JdbcUtil.update(...)` runs the CAS `UPDATE` inside
`connections.run(...)` with that default retry (`JdbcUtil.java:533-561`). The
statement is not idempotent: once applied, a replay updates 0 rows.
3. `JdbcTableOperations.updateTable` turns anything other than 1 row into
`CommitFailedException` (`JdbcTableOperations.java:146-163`) without checking
whether `newMetadataLocation` is already the table's current (or a historical)
metadata location.
4. `CommitFailedException implements CleanableFailure`;
`TableOperations.requireStrictCleanup()` defaults to true and JDBC does not
override it. `SnapshotProducer.commit()` then calls `cleanAll()`
(`SnapshotProducer.java:536-540, 610-616`), and `BaseTransaction` does the same
through `cleanUp()` / `cleanAllUpdates()` (`BaseTransaction.java:380-385,
415-432`). Both delete through `deleteFile` -> the table `FileIO`.
Other catalogs already reconcile this case: `HiveTableOperations` (`:375`),
`GlueTableOperations` (`:174`), `DynamoDbTableOperations` and
`BigQueryTableOperations` call `checkCommitStatus`/`checkCommitStatusStrict`
from `BaseMetastoreOperations` when the outcome is uncertain. The JDBC table
and view operations do not.
**Reproduction.** A small PostgreSQL-protocol-aware TCP proxy sits between
the catalog and the server. Once armed, it forwards the catalog's `UPDATE
iceberg_tables SET metadata_location ...`, withholds every backend message up
to and including ReadyForQuery, and closes both sockets. PostgreSQL has
committed the row; the client only sees a dropped connection. Full sources
(about 350 lines, Gradle project, no object store needed) are in the collapsed
section at the end. Run:
```
docker run -d --name iceberg-repro-pg -e POSTGRES_PASSWORD=pw -p
127.0.0.1:55432:5432 postgres:17-alpine
gradle installDist
build/install/iceberg-repro/bin/iceberg-repro 55432 2>&1 | grep '^RESULT'
```
Output on stock 1.12.0 (ids vary per run):
```
RESULT plain_retries_default: responseDropped=true | commit returned
normally | table: snapshots=1, current manifest list exists=true, scan OK, data
files=1 | deleteFile calls:
[snap-429960218544967894-2-b4186a2a-1e7d-40a5-9931-b097c60e6bc4.avro]
RESULT plain_retries_0: responseDropped=true | threw CommitFailedException:
Failed to update table ns.plain_retries_0 from catalog c | table: snapshots=1,
current manifest list exists=false, scan FAILED NotFoundException: No in-memory
file found for location:
mem://wh/ns/plain_retries_0/metadata/snap-1137813634021871497-1-a9635f87-9362-4b3f-9683-f1e347777bfc.avro
| deleteFile calls:
[snap-1137813634021871497-1-a9635f87-9362-4b3f-9683-f1e347777bfc.avro,
a9635f87-9362-4b3f-9683-f1e347777bfc-m0.avro]
RESULT txn_retries_default: responseDropped=true | commit returned normally
| table: snapshots=1, current manifest list exists=true, scan OK, data files=1
| deleteFile calls:
[snap-8670333551669072922-2-8668fa5b-da14-45b6-95d5-43bfaf649abc.avro]
RESULT txn_retries_0: responseDropped=true | threw CommitFailedException:
Failed to update table ns.txn_retries_0 from catalog c | table: snapshots=1,
current manifest list exists=false, scan FAILED NotFoundException: No in-memory
file found for location:
mem://wh/ns/txn_retries_0/metadata/snap-9194260192876454534-1-03961be1-29bc-4877-9a09-22de841777f3.avro
| deleteFile calls:
[snap-9194260192876454534-1-03961be1-29bc-4877-9a09-22de841777f3.avro,
03961be1-29bc-4877-9a09-22de841777f3-m0.avro]
RESULT concurrent_writer_default_retries: responseDropped=true | writer 1
commit returned normally | after both writers returned: snapshots in
metadata=2, data files visible on main=1 (expected 2: writer 1 and the
concurrent writer)
```
The stack of the false failure (retries=0, transaction case):
```
org.apache.iceberg.exceptions.CommitFailedException: Failed to update table
ns.txn_retries_0 from catalog c
at
org.apache.iceberg.jdbc.JdbcTableOperations.updateTable(JdbcTableOperations.java:160)
at
org.apache.iceberg.jdbc.JdbcTableOperations.doCommit(JdbcTableOperations.java:115)
at
org.apache.iceberg.BaseMetastoreTableOperations.commit(BaseMetastoreTableOperations.java:127)
at
org.apache.iceberg.BaseTransaction.lambda$commitSimpleTransaction$1(BaseTransaction.java:371)
at
org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry(Tasks.java:416)
...
```
**Expected.** A commit that the database applied is reported as succeeded,
or at worst as `CommitStateUnknownException` (not cleanable). Files referenced
by the table's current metadata are never deleted, and a concurrent writer's
committed snapshot is never rewound.
**Relation to existing work.** #18402 / #18407 describe the same cleanup
consequence (an applied commit treated as a cleanable failure, then
`cleanAll()`), triggered on the REST/Flink side; the comment on #18407 about a
concurrent-writer rewind is exactly the fifth scenario above, reproduced here
on the JDBC catalog. #18407's guard would stop the file deletion but not the
false `CommitFailedException` or the rewind, so the two are complementary.
#17828 / #17829 (MySQL `SQLRecoverableException`) would widen the same replay
surface.
**Proposed fix (prototype verified).** In `JdbcTableOperations.updateTable`,
when the UPDATE affects 0 rows, call the existing
`checkCommitStatusStrict(newMetadataLocation, metadata)` before throwing. The
new metadata location is unique per attempt, so finding it as the current
location or in the metadata log proves the UPDATE landed:
* `SUCCESS`: return as committed (log at INFO);
* `UNKNOWN`: throw `CommitStateUnknownException` (not cleanable);
* `FAILURE`: `CommitFailedException` as today.
`JdbcViewOperations.updateView` (`:159-168`) has the same shape and needs
the same change. With the patch (diff in the collapsed section), the same five
scenarios give:
```
RESULT plain_retries_default: ... commit returned normally | ... scan OK,
data files=1 | deleteFile calls: []
RESULT plain_retries_0: ... commit returned normally | ... scan OK, data
files=1 | deleteFile calls: []
RESULT txn_retries_default: ... commit returned normally | ... scan OK, data
files=1 | deleteFile calls: []
RESULT txn_retries_0: ... commit returned normally | ... scan OK, data
files=1 | deleteFile calls: []
RESULT concurrent_writer_default_retries: ... writer 1 commit returned
normally | ... snapshots in metadata=2, data files visible on main=2 (expected
2: writer 1 and the concurrent writer)
```
A variant that only compares the table's *current* metadata location fixes
the first four scenarios but not the concurrent-writer one, so the
history-aware check matters. Cost: one extra refresh on the failure path only.
I am happy to open the PR (table + view operations, plus a `TestJdbcCatalog`
case that executes the UPDATE and then throws an `SQLException` with state
`08006` once, with and without retries, and with a second writer in between),
and to coordinate with #18407.
<details>
<summary>Reproduction sources (Gradle project, JDK 21)</summary>
`settings.gradle.kts`
```kotlin
rootProject.name = "iceberg-repro"
```
`build.gradle.kts`
```kotlin
plugins { java; application }
repositories { mavenCentral() }
java { toolchain { languageVersion = JavaLanguageVersion.of(21) } }
dependencies {
implementation("org.apache.iceberg:iceberg-core:1.12.0")
implementation("org.apache.iceberg:iceberg-api:1.12.0")
runtimeOnly("org.postgresql:postgresql:42.7.14")
runtimeOnly("org.slf4j:slf4j-simple:2.0.17")
}
application { mainClass = "repro.LostAck" }
```
`src/main/java/repro/LostAck.java`
```java
package repro;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.jdbc.JdbcCatalog;
import org.apache.iceberg.types.Types;
public final class LostAck {
static Map<String, String> props(int port) {
Map<String, String> p = new HashMap<>();
p.put("uri", "jdbc:postgresql://127.0.0.1:" + port +
"/postgres?sslmode=disable&gssEncMode=disable");
p.put("jdbc.user", "postgres");
p.put("jdbc.password", "pw");
p.put("jdbc.schema-version", "V1");
p.put("jdbc.init-catalog-tables", "true");
p.put("warehouse", "mem://wh");
p.put("io-impl", RecordingFileIO.class.getName());
return p;
}
public static void main(String[] args) throws Exception {
int pgPort = Integer.parseInt(args[0]);
try (PgProxy proxy = new PgProxy(pgPort)) {
JdbcCatalog catalog = new JdbcCatalog();
catalog.initialize("c", props(proxy.port()));
catalog.createNamespace(org.apache.iceberg.catalog.Namespace.of("ns"));
Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.LongType.get()));
int i = 0;
for (boolean txn : new boolean[] {false, true}) {
for (String retries : new String[] {"default(4)", "0"}) {
String name = (txn ? "txn" : "plain") + "_retries_" +
(retries.equals("0") ? "0" : "default");
run(catalog, proxy, schema, name, txn,
retries.equals("0"), i++);
}
}
concurrentWriter(catalog, proxy, schema);
catalog.close();
}
}
static void run(JdbcCatalog catalog, PgProxy proxy, Schema schema,
String name, boolean txn, boolean noRetry, int n)
throws Exception {
TableIdentifier id = TableIdentifier.of("ns", name);
Map<String, String> tp = new HashMap<>();
if (noRetry) {
tp.put("commit.retry.num-retries", "0");
}
Table t = catalog.createTable(id, schema,
PartitionSpec.unpartitioned(), tp);
DataFile df = DataFiles.builder(PartitionSpec.unpartitioned())
.withPath("mem://wh/data/" + name + "-1.parquet")
.withFormat(FileFormat.PARQUET)
.withFileSizeInBytes(10)
.withRecordCount(5)
.build();
RecordingFileIO.DELETED.clear();
proxy.drops = 0;
proxy.armed.set(true);
String outcome;
try {
if (txn) {
Transaction tx = t.newTransaction();
tx.newAppend().appendFile(df).commit();
tx.commitTransaction();
} else {
AppendFiles a = t.newAppend().appendFile(df);
a.commit();
}
outcome = "commit returned normally";
} catch (Exception e) {
outcome = "threw " + e.getClass().getSimpleName() + ": " +
firstLine(e);
}
boolean dropped = proxy.drops > 0;
proxy.armed.set(false);
// Inspect with a fresh catalog connected straight to PostgreSQL (no
proxy).
JdbcCatalog fresh = new JdbcCatalog();
fresh.initialize("c", props(proxy.upstream()));
String state;
try {
Table r = fresh.loadTable(id);
Snapshot cur = r.currentSnapshot();
if (cur == null) {
state = "no current snapshot";
} else {
boolean listExists;
try {
listExists =
r.io().newInputFile(cur.manifestListLocation()).exists();
} catch (RuntimeException e) {
listExists = false; // InMemoryFileIO throws
NotFoundException for a missing object
}
String files;
try {
int c = 0;
try (var tasks = r.newScan().planFiles()) {
for (var ignored : tasks) {
c++;
}
}
files = "scan OK, data files=" + c;
} catch (Exception e) {
files = "scan FAILED " + e.getClass().getSimpleName() +
": " + firstLine(e);
}
state = "snapshots=" + countSnapshots(r) + ", current
manifest list exists=" + listExists + ", " + files;
}
} catch (Exception e) {
state = "load FAILED " + e;
}
fresh.close();
System.out.println("RESULT " + name + ": responseDropped=" + dropped
+ " | " + outcome + " | table: " + state
+ " | deleteFile calls: " + RecordingFileIO.DELETED);
}
/** Lost acknowledgement, then another writer commits on top before the
first writer's retry (default retries). */
static void concurrentWriter(JdbcCatalog catalog, PgProxy proxy, Schema
schema) throws Exception {
TableIdentifier id = TableIdentifier.of("ns",
"concurrent_default_retries");
Table t = catalog.createTable(id, schema);
JdbcCatalog other = new JdbcCatalog();
other.initialize("c", props(proxy.upstream()));
proxy.onDrop = () -> {
Table o = other.loadTable(id);
o.newFastAppend().appendFile(file("concurrent-B")).commit();
};
proxy.armed.set(true);
String outcome;
try {
t.newAppend().appendFile(file("writer1-A")).commit();
outcome = "commit returned normally";
} catch (Exception e) {
outcome = "threw " + e.getClass().getSimpleName();
}
proxy.armed.set(false);
proxy.onDrop = null;
Table r = other.loadTable(id);
int files = 0;
try (var tasks = r.newScan().planFiles()) {
for (var task : tasks) {
System.out.println(" data file in current table state: " +
task.file().location());
files++;
}
}
other.close();
System.out.println("RESULT concurrent_writer_default_retries:
responseDropped=" + (proxy.drops > 0) + " | writer 1 " + outcome
+ " | after both writers returned: snapshots in metadata=" +
countSnapshots(r) + ", data files visible on main=" + files
+ " (expected 2: writer 1 and the concurrent writer)");
}
static DataFile file(String name) {
return DataFiles.builder(PartitionSpec.unpartitioned())
.withPath("mem://wh/data/" + name + ".parquet")
.withFormat(FileFormat.PARQUET)
.withFileSizeInBytes(10)
.withRecordCount(5)
.build();
}
static int countSnapshots(Table t) {
int c = 0;
for (Snapshot ignored : t.snapshots()) {
c++;
}
return c;
}
static String firstLine(Throwable e) {
String m = String.valueOf(e.getMessage());
int nl = m.indexOf('\n');
return nl > 0 ? m.substring(0, nl) : m;
}
}
```
`src/main/java/repro/PgProxy.java`
```java
package repro;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* TCP proxy that forwards PostgreSQL traffic and, once armed, lets the
first catalog
* "UPDATE iceberg_tables SET metadata_location" reach the server, then
swallows the whole response and closes
* both sockets. PostgreSQL has committed the row; the client only sees a
dropped connection.
*/
final class PgProxy implements AutoCloseable {
private static final byte[] CAS =
"UPDATE iceberg_tables SET
metadata_location".getBytes(StandardCharsets.US_ASCII);
private final ServerSocket server;
private final int upstreamPort;
final AtomicBoolean armed = new AtomicBoolean();
volatile int drops;
/** Runs after PostgreSQL committed the UPDATE and before the client
learns anything. */
volatile Runnable onDrop;
PgProxy(int upstreamPort) throws IOException {
this.upstreamPort = upstreamPort;
this.server = new ServerSocket(0, 50,
InetAddress.getLoopbackAddress());
Thread t = new Thread(this::accept, "proxy-accept");
t.setDaemon(true);
t.start();
}
int port() {
return server.getLocalPort();
}
int upstream() {
return upstreamPort;
}
private void accept() {
while (!server.isClosed()) {
try {
Socket c = server.accept();
Socket u = new Socket(InetAddress.getLoopbackAddress(),
upstreamPort);
boolean[] casSent = {false};
daemon(() -> c2s(c, u, casSent));
daemon(() -> s2c(c, u, casSent));
} catch (IOException e) {
return;
}
}
}
private static void daemon(Runnable r) {
Thread t = new Thread(r);
t.setDaemon(true);
t.start();
}
private void c2s(Socket c, Socket u, boolean[] casSent) {
try (InputStream in = c.getInputStream(); OutputStream out =
u.getOutputStream()) {
byte[] buf = new byte[65536];
int n;
while ((n = in.read(buf)) > 0) {
if (armed.get() && !casSent[0] && indexOf(buf, n) >= 0) {
casSent[0] = true;
}
out.write(buf, 0, n);
out.flush();
}
} catch (IOException ignored) {
// closed
} finally {
close(c, u);
}
}
private void s2c(Socket c, Socket u, boolean[] casSent) {
try (DataInputStream in = new DataInputStream(u.getInputStream());
OutputStream out = c.getOutputStream()) {
while (true) {
int type = in.read();
if (type < 0) {
return;
}
int len = in.readInt();
byte[] body = new byte[len - 4];
in.readFully(body);
if (casSent[0] && armed.get()) {
if (type == 'Z') { // ReadyForQuery: the statement has
been committed by the server
armed.set(false);
drops++;
Runnable r = onDrop;
if (r != null) {
r.run();
}
return;
}
continue; // withhold ParseComplete, BindComplete,
CommandComplete "UPDATE 1", ...
}
out.write(type);
out.write(new byte[] {(byte) (len >>> 24), (byte) (len >>>
16), (byte) (len >>> 8), (byte) len});
out.write(body);
out.flush();
}
} catch (IOException ignored) {
// closed
} finally {
close(c, u);
}
}
private static int indexOf(byte[] hay, int n) {
outer:
for (int i = 0; i + CAS.length <= n; i++) {
for (int j = 0; j < CAS.length; j++) {
if (hay[i + j] != CAS[j]) {
continue outer;
}
}
return i;
}
return -1;
}
private static void close(Socket a, Socket b) {
try { a.close(); } catch (IOException ignored) { }
try { b.close(); } catch (IOException ignored) { }
}
@Override
public void close() throws IOException {
server.close();
}
}
```
`src/main/java/repro/RecordingFileIO.java`
```java
package repro;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.iceberg.inmemory.InMemoryFileIO;
/** InMemoryFileIO that records every deleteFile call. */
public class RecordingFileIO extends InMemoryFileIO {
static final List<String> DELETED = new CopyOnWriteArrayList<>();
@Override
public void deleteFile(String location) {
DELETED.add(location.substring(location.lastIndexOf('/') + 1));
super.deleteFile(location);
}
}
```
</details>
<details>
<summary>Prototype patch against the 1.12.0 sources</summary>
```diff
--- a/core/src/main/java/org/apache/iceberg/jdbc/JdbcTableOperations.java
+++ b/core/src/main/java/org/apache/iceberg/jdbc/JdbcTableOperations.java
@@ -32,6 +32,7 @@
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.io.FileIO;
@@ -112,7 +113,7 @@
String oldMetadataLocation = base.metadataFileLocation();
// Start atomic update
LOG.debug("Committing existing table: {}", tableName());
- updateTable(newMetadataLocation, oldMetadataLocation);
+ updateTable(newMetadataLocation, oldMetadataLocation, metadata);
} else {
// table not exists create it
LOG.debug("Committing new table: {}", tableName());
@@ -143,7 +144,8 @@
}
}
- private void updateTable(String newMetadataLocation, String
oldMetadataLocation)
+ private void updateTable(
+ String newMetadataLocation, String oldMetadataLocation, TableMetadata
metadata)
throws SQLException, InterruptedException {
int updatedRecords =
JdbcUtil.updateTable(
@@ -156,9 +158,23 @@
if (updatedRecords == 1) {
LOG.debug("Successfully committed to existing table: {}",
tableIdentifier);
- } else {
- throw new CommitFailedException(
- "Failed to update table %s from catalog %s", tableIdentifier,
catalogName);
+ return;
+ }
+
+ // The UPDATE is a compare-and-swap. Zero rows means either another
writer won, or the connection
+ // dropped after the database applied this UPDATE and the pool replayed
it. The new metadata
+ // location is unique to this attempt, so finding it in the table's
history proves it was applied.
+ switch (checkCommitStatusStrict(newMetadataLocation, metadata)) {
+ case SUCCESS:
+ LOG.info("Commit to {} was already applied, treating it as
successful", tableIdentifier);
+ return;
+ case UNKNOWN:
+ throw new CommitStateUnknownException(
+ new CommitFailedException(
+ "Failed to update table %s from catalog %s",
tableIdentifier, catalogName));
+ default:
+ throw new CommitFailedException(
+ "Failed to update table %s from catalog %s", tableIdentifier,
catalogName);
}
}
```
</details>
### Willingness to contribute
- [x] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from
the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
