gardenia opened a new issue, #18902:
URL: https://github.com/apache/iceberg/issues/18902
### Apache Iceberg version
1.12.0 (latest release)
### Query engine
Spark
### Please describe the bug 🐞
## Describe the bug
If the connection fails after a commit request has been sent to a REST
catalog (no response, a
connection reset, a read timeout), the Java client reports an ordinary
failure:
```
org.apache.iceberg.exceptions.RESTException: Error occurred while processing
POST request
Caused by: org.apache.hc.core5.http.NoHttpResponseException: 127.0.0.1:43911
failed to respond
```
The server may well have applied the commit. In that case the outcome is
unknown, just as for a
5xx response, but the client doesn't say so:
- It is not a `CommitStateUnknownException`, so callers have no warning that
the commit may have
landed. An application or scheduler that retries the failed operation
writes the data twice.
- `RESTTableOperations.commit` never attempts the reconciliation it already
does for 5xx
responses, where it reloads the table and checks whether the snapshot it
added is there.
For comparison, the same commit answered with a 504 is reconciled: Spark
reports success, with one
copy of the data.
## Cause
`HTTPClient.execute` wraps any `IOException` in a plain `RESTException`, so
a transport failure
never reaches `ErrorHandlers.tableCommitHandler()`, which maps
500/502/503/504 on a commit to
`CommitStateUnknownException`. `RESTTableOperations` already carries a note
about this:
```java
// TODO: ensure that the HTTP client lib passes HTTP client errors to the
error handler
```
`RESTSessionCatalog.commitTransaction` (multi-table transactions) has the
same problem.
## To reproduce
With Iceberg's own reference catalog (`apache/iceberg-rest-fixture`) and a
small proxy between Spark
and the catalog that forwards one table commit, lets the catalog apply it,
and then closes the
connection instead of relaying the response:
```python
"""Forwards HTTP to a REST catalog. After GET /__arm, the next table commit
is
forwarded (so the catalog applies it) but its response is never sent: the
connection is closed instead, as a network failure would."""
import http.client, http.server, re, sys, threading
LISTEN, UPSTREAM = int(sys.argv[1]), int(sys.argv[2])
armed = threading.Event()
COMMIT = re.compile(r"/namespaces/[^/]+/tables/[^/]+$")
SKIP = {"host", "connection", "content-length", "transfer-encoding"}
class Handler(http.server.BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def forward(self):
if self.path == "/__arm":
armed.set()
self.send_response(200); self.send_header("Content-Length",
"0"); self.end_headers()
return
n = int(self.headers.get("Content-Length") or 0)
body = self.rfile.read(n) if n else None
upstream = http.client.HTTPConnection("127.0.0.1", UPSTREAM)
upstream.request(self.command, self.path, body, {k: v for k, v in
self.headers.items() if k.lower() not in SKIP})
resp = upstream.getresponse()
data = resp.read()
if self.command == "POST" and COMMIT.search(self.path.split("?")[0])
and armed.is_set():
armed.clear()
print(f"proxy: catalog answered {resp.status} to POST
{self.path}; dropping the connection instead", flush=True)
self.close_connection = True
return
self.send_response(resp.status)
for k, v in resp.getheaders():
if k.lower() not in SKIP:
self.send_header(k, v)
self.send_header("Content-Length", str(len(data)))
self.end_headers()
if self.command != "HEAD":
self.wfile.write(data)
do_GET = do_POST = do_DELETE = do_HEAD = do_PUT = forward
def log_message(self, *args):
pass
http.server.ThreadingHTTPServer(("127.0.0.1", LISTEN),
Handler).serve_forever()
```
```sh
docker run -d --network host -e CATALOG_WAREHOUSE=file:///tmp/wh/ -v
/tmp/wh:/tmp/wh apache/iceberg-rest-fixture
python3 proxy.py 18291 8181
```
Then, in Spark 3.5 with `iceberg-spark-runtime-3.5_2.12-1.12.0`, with
catalog `via` pointing at the
proxy (`http://127.0.0.1:18291`) and catalog `direct` at the catalog
(`http://127.0.0.1:8181`):
```sql
CREATE NAMESPACE via.x;
CREATE TABLE via.x.t (id bigint) USING iceberg;
INSERT INTO via.x.t VALUES (1);
-- GET http://127.0.0.1:18291/__arm
INSERT INTO via.x.t VALUES (2); -- RESTException: Error occurred while
processing POST request
SELECT count(*) FROM direct.x.t WHERE id = 2; -- 1: the commit landed
INSERT INTO via.x.t VALUES (2); -- the obvious retry succeeds
SELECT count(*) FROM direct.x.t WHERE id = 2; -- 2: duplicated
```
The proxy logs that the catalog answered the dropped commit with a 200.
Other engines and operations behave the same way. Checked with another REST
catalog that can inject
the same fault:
| operation | engine reports | did the commit land? |
|---|---|---|
| Spark `INSERT`, catalog answers 504 after applying it | success
(reconciled) | yes, one copy |
| Spark `INSERT`, connection closed after applying it | `RESTException` |
yes |
| Trino `INSERT`, connection closed after applying it | `Failed to commit
the transaction during insert: Error occurred while processing POST request` |
yes |
| Trino `DELETE`, connection closed after applying it | `Failed to commit
the transaction during write: Error occurred while processing POST request` |
yes |
The same holds for Spark `rewrite_data_files`, CTAS, `CREATE OR REPLACE
TABLE … AS SELECT` and
`DELETE`, and Trino CTAS and `optimize`. In every case the engines kept the
data files, so the
tables themselves stay consistent; the risk is a retry.
## Related
- #6778 reported this in 2023. It was closed as fixed by #8397 and #8599,
which made engines clean up
only after failures marked `CleanableFailure`. That stopped the data loss:
the engines above all
keep their files. But the commit is still reported as an ordinary failure,
and still isn't
reconciled, which is what this issue is about. The concern raised there
was converting every
`IOException` in `HTTPClient`; the fix proposed below applies only to
commit requests and keeps a
failure to connect as an ordinary failure.
- #12792 hit the same cause in `rewrite_manifests` (`RESTException` caused by
`SocketTimeoutException: Read timed out`), and the discussion there agreed
an `IOException` on a
commit should be a `CommitStateUnknownException`. No fix followed for that
part.
- #14320 added the reconciliation for 5xx that this would extend to
transport failures.
## Proposed fix
Report a transport failure on a commit request as
`CommitStateUnknownException`, so that it is
handled like a 5xx: snapshot-add-only commits are reconciled, and anything
else is reported as
unknown rather than failed. A `ConnectException` proves the request was
never sent, so it can stay
an ordinary failure. This applies to `RESTTableOperations.commit` and
`RESTSessionCatalog.commitTransaction`.
With that change, the Spark `INSERT` whose connection was closed after the
commit reports success
with one copy, the same as for the 504, against both the reference catalog
and another REST catalog.
The change applies cleanly to `main`, where the REST client tests pass.
### 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]
