Add idempotency adapter and E2E coverage #14773

huaxingao · 2025-12-05T23:45:31Z

This is the 4th PR for Idempotency Key. Introduce test-only idempotency in RESTCatalogAdapter:

in-memory store keyed by Idempotency-Key,
cached-success replay
TTL expiry
optional “503 after success” simulation.

Add E2E tests:

testIdempotentDuplicateCreateReturnsCached
testIdempotencyKeyLifetimeExpiredTreatsAsNew
testIdempotentCreateReplayAfterSimulated503

First PR: #14649
SecondPR: #14700
Third PR: #14740

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

nastra · 2025-12-10T14:01:05Z

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

+      long now = System.currentTimeMillis();
+      boolean expired =
+          existing.status == IdempotencyEntry.Status.FINALIZED
+              && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;


can we use Instant.now().minusMillis(existing.firstSeenMillis) here as that is more readable when adding isBefore / isAfter

you could also make this a function on the entry itself, so that you'd only call if(!entry.expired()) here

Fixed. Thanks!

nastra · 2025-12-10T14:06:17Z

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

+  }
+
+  // Test hooks/configuration for idempotency behavior
+  public static void simulate503OnFirstSuccessForKey(String key) {


I'm not really a fan of such a way to configure this. We should explore doing this similarly how remote scan planning sets the planning behavior for particular tests

Thanks for the suggestion. I removed the static hook from CatalogHandlers and mirrored the remote planning pattern:

Added a per-adapter IdempotencyBehavior in RESTCatalogAdapter and apply it post‑success in execute(...).

Tests configure behavior via the adapter (e.g., adapterForRESTServer.simulate503OnFirstSuccessForKey(key)), while routes still call CatalogHandlers.withIdempotency(...).

nastra · 2025-12-10T14:10:58Z

core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java

+    AuthSession httpSession = am.initSession(httpBase, conf);
+    RESTClient http = httpBase.withAuthSession(httpSession);
+
+    CreateTableRequest req =


seems a bit weird to create the request here. Can we not mock stuff as we do in a bunch of other tests and verify that the correct requests were sent?

I’ve added a reusable verify helper and now assert the request shape (method/path/headers, including Idempotency‑Key) after the POSTs. I’m keeping a real CreateTableRequest in the E2E tests because it’s needed to exercise serialization and replay.

nastra · 2025-12-10T14:11:12Z

core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java

+    return new IdempotentCreateEnv(ns, ident, headers, http, req);
+  }
+
+  private static class IdempotentCreateEnv {


do we really need this wrapper class?

I removed this wrapped class

huaxingao · 2025-12-15T18:36:37Z

@nastra @amogh-jahagirdar @singhpk234 Could you please take a look when you have a moment? Thanks!

singhpk234 · 2025-12-20T00:12:46Z

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

+    simulate503OnFirstSuccessKeys.add(key);
+  }
+
  private static OAuthTokenResponse handleOAuthRequest(Object body) {


can we wrap this to plan-api endpoints too ?

Good point. Plan endpoints aren’t wrapped with withIdempotency yet. I’ll follow up with a separate PR to add Idempotency-Key support for the plan APIs (wrap handlers + add tests) and we can extend the simulation coverage there.

singhpk234 · 2025-12-20T00:14:55Z

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

  }

+  /** Test helper to simulate a transient 503 after the first successful mutation for a key. */
+  public void simulate503OnFirstSuccessForKey(String key) {


nit : should we make it generic like simulate status code or something ?

Good suggestion. I kept it 503-specific since the test hook is only used to exercise retry behavior for transient 503/commit-state-unknown; happy to generalize later if we add tests needing other codes.

singhpk234 · 2025-12-20T02:29:27Z

core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java

+    String key = "idemp-drop-void";
+    Namespace ns = Namespace.of("ns_void");
+    TableIdentifier ident = TableIdentifier.of(ns, "t_void");
+    restCatalog.createNamespace(ns, ImmutableMap.of());
+    Pair<RESTClient, Map<String, String>> httpAndHeaders = httpAndHeaders(key);
+    RESTClient http = httpAndHeaders.first();
+    Map<String, String> headers = httpAndHeaders.second();


i wonder if all this can be abstracted in a utils which takes key, namespace and table_name ? wdyt ?

Thanks for the comment! I have added a smaller helper method for the common code.

geruh · 2025-12-23T19:30:41Z

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

+    String key = keyHeader.get().value();
+
+    // check existing entry and TTL
+    IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);


This is rare in this use case but maybe we should use computeIfAbsent here to avoid any race conditions get and put issues

Good call — I updated this to avoid the get/remove/put race by using a single atomic map operation. I now use IDEMPOTENCY_STORE.compute(...) to either reuse the current entry or replace it if missing/expired, and use a per-call isLeader flag so only the thread that created the new entry executes the action while others wait and replay the finalized result. I used compute (instead of computeIfAbsent) because we also need to replace expired entries.

singhpk234

Mostly LGTM, thanks @huaxingao left minor comments

singhpk234 · 2026-01-06T01:08:32Z

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

+
+    String key = keyHeader.get().value();
+
+    AtomicBoolean isLeader = new AtomicBoolean(false);


whats a Leader ?

“Leader” here means the request thread that won the IDEMPOTENCY_STORE.compute(...) and created (or replaced) the IN_PROGRESS entry for this Idempotency-Key. That thread executes the action and finalizes the entry; all other concurrent requests for the same key are “followers” that wait on the latch and then replay the finalized result/error.

singhpk234 · 2026-01-06T01:24:41Z

core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java

+          .hasMessageContaining(ident.toString());
+
+      // Clean up
+      restCatalog.dropTable(ident);


do we need explicit drop ?

Good catch. This is not needed. Removed.

singhpk234 · 2026-01-06T02:11:05Z

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

+    try {
+      idempotencyLifetimeMillis = Duration.parse(isoDuration).toMillis();
+    } catch (Exception e) {
+      // ignore parse errors; keep default


sorry i might have missed this in first pass, this is public imho we should throw illegalArgs ?

Good catch. I made this test-only (@VisibleForTesting + package-private) and changed it to throw IllegalArgumentException on invalid ISO durations rather than silently ignoring parse errors.

singhpk234 · 2026-01-06T02:14:13Z

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java


  private AuthSession authSession = AuthSession.EMPTY;
  private PlanningBehavior planningBehavior;
+  private final java.util.Set<String> simulate503OnFirstSuccessKeys = Sets.newConcurrentHashSet();


nit: can we import this instead of inline ?

Fixed. Thanks

github-actions bot added the core label Dec 5, 2025

huaxingao requested review from nastra and singhpk234 December 6, 2025 01:28

nastra reviewed Dec 8, 2025

View reviewed changes

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java Outdated Show resolved Hide resolved

nastra reviewed Dec 10, 2025

View reviewed changes

nastra requested a review from amogh-jahagirdar December 10, 2025 14:12

huaxingao force-pushed the idempotency_4 branch 2 times, most recently from 47aae84 to a8a0a16 Compare December 15, 2025 18:34

singhpk234 reviewed Dec 20, 2025

View reviewed changes

geruh reviewed Dec 23, 2025

View reviewed changes

huaxingao added 4 commits January 2, 2026 14:27

rebase

76ad9fe

resolve conflicts

e16d858

address comments

2f59010

rebase to upstream

4a81530

huaxingao force-pushed the idempotency_4 branch from a8a0a16 to 4a81530 Compare January 2, 2026 22:34

singhpk234 reviewed Jan 6, 2026

View reviewed changes

address comments

c3e5460


		String key = keyHeader.get().value();

		AtomicBoolean isLeader = new AtomicBoolean(false);

Add idempotency adapter and E2E coverage #14773

Are you sure you want to change the base?

Add idempotency adapter and E2E coverage #14773

Uh oh!

Conversation

huaxingao commented Dec 5, 2025

Uh oh!

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!

huaxingao commented Dec 15, 2025

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!

singhpk234 left a comment

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants