You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "huaxingao (via GitHub)" <gi...@apache.org> on 2025/12/05 23:45:34 UTC
[PR] Add idempotency adapter and E2E coverage [iceberg]
huaxingao opened a new pull request, #14773:
URL: https://github.com/apache/iceberg/pull/14773
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: https://github.com/apache/iceberg/pull/14649
SecondPR: https://github.com/apache/iceberg/pull/14700
Third PR: https://github.com/apache/iceberg/pull/14740
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2606815069
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3345,6 +3470,73 @@ public <T extends RESTResponse> T execute(
return local;
}
+ private IdempotentCreateEnv prepareIdempotentCreateEnv(
+ String namespaceName, String tableName, String idempotencyKey) {
+ Namespace ns = Namespace.of(namespaceName);
+ TableIdentifier ident = TableIdentifier.of(ns, tableName);
+ Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
+
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+
+ Map<String, String> headers =
+ ImmutableMap.of(
+ RESTUtil.IDEMPOTENCY_KEY_HEADER,
+ idempotencyKey,
+ "Authorization",
+ "Bearer client-credentials-token:sub=user",
+ "test-header",
+ "test-value");
+
+ Map<String, String> conf =
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ httpServer.getURI().toString(),
+ HTTPClient.REST_SOCKET_TIMEOUT_MS,
+ "600000",
+ HTTPClient.REST_CONNECTION_TIMEOUT_MS,
+ "600000",
+ "header.test-header",
+ "test-value");
+ RESTClient httpBase =
+ HTTPClient.builder(conf)
+ .uri(conf.get(CatalogProperties.URI))
+ .withHeaders(RESTUtil.configHeaders(conf))
+ .build();
+ AuthManager am = AuthManagers.loadAuthManager("test", conf);
+ AuthSession httpSession = am.initSession(httpBase, conf);
+ RESTClient http = httpBase.withAuthSession(httpSession);
+
+ CreateTableRequest req =
Review Comment:
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?
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3345,6 +3470,73 @@ public <T extends RESTResponse> T execute(
return local;
}
+ private IdempotentCreateEnv prepareIdempotentCreateEnv(
+ String namespaceName, String tableName, String idempotencyKey) {
+ Namespace ns = Namespace.of(namespaceName);
+ TableIdentifier ident = TableIdentifier.of(ns, tableName);
+ Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
+
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+
+ Map<String, String> headers =
+ ImmutableMap.of(
+ RESTUtil.IDEMPOTENCY_KEY_HEADER,
+ idempotencyKey,
+ "Authorization",
+ "Bearer client-credentials-token:sub=user",
+ "test-header",
+ "test-value");
+
+ Map<String, String> conf =
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ httpServer.getURI().toString(),
+ HTTPClient.REST_SOCKET_TIMEOUT_MS,
+ "600000",
+ HTTPClient.REST_CONNECTION_TIMEOUT_MS,
+ "600000",
+ "header.test-header",
+ "test-value");
+ RESTClient httpBase =
+ HTTPClient.builder(conf)
+ .uri(conf.get(CatalogProperties.URI))
+ .withHeaders(RESTUtil.configHeaders(conf))
+ .build();
+ AuthManager am = AuthManagers.loadAuthManager("test", conf);
+ AuthSession httpSession = am.initSession(httpBase, conf);
+ RESTClient http = httpBase.withAuthSession(httpSession);
+
+ CreateTableRequest req =
+ CreateTableRequest.builder()
+ .withName(ident.name())
+ .withSchema(schema)
+ .withPartitionSpec(PartitionSpec.unpartitioned())
+ .build();
+
+ return new IdempotentCreateEnv(ns, ident, headers, http, req);
+ }
+
+ private static class IdempotentCreateEnv {
Review Comment:
do we really need this wrapper class?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2658490063
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -122,6 +124,11 @@ public RESTCatalogAdapter(Catalog catalog) {
this.asViewCatalog = catalog instanceof ViewCatalog ? (ViewCatalog) catalog : null;
}
+ /** Test helper to simulate a transient 503 after the first successful mutation for a key. */
+ public void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
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.
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -122,6 +124,11 @@ public RESTCatalogAdapter(Catalog catalog) {
this.asViewCatalog = catalog instanceof ViewCatalog ? (ViewCatalog) catalog : null;
}
+ /** Test helper to simulate a transient 503 after the first successful mutation for a key. */
+ public void simulate503OnFirstSuccessForKey(String key) {
+ simulate503OnFirstSuccessKeys.add(key);
+ }
+
private static OAuthTokenResponse handleOAuthRequest(Object body) {
Review Comment:
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.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#issuecomment-3746038302
Thanks @amogh-jahagirdar @geruh @nastra @singhpk234 for your review! Are there any more comments? If not, I plan to merge this PR in two days.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2608646989
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
Review Comment:
Fixed. Thanks!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2606779973
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
Review Comment:
can we use `Instant.now().minusMillis(existing.firstSeenMillis)` here as that is more readable when adding `isBefore` / `isAfter`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2597414587
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -557,6 +568,19 @@ protected <T extends RESTResponse> T execute(
vars.putAll(request.queryParameters());
vars.putAll(routeAndVars.second());
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ request.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ boolean isMutation =
+ request.method() == HTTPMethod.POST || request.method() == HTTPMethod.DELETE;
+ if (isMutation && keyHeader.isPresent()) {
+ T response =
+ handleIdempotentMutation(
Review Comment:
our reference implementation for a REST catalog all lives in `CatalogHandlers` and the adapter is really just a wrapper that forwards the queries to that. Can you please move the ref impl there?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2668710925
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +115,148 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ AtomicBoolean isLeader = new AtomicBoolean(false);
+ IdempotencyEntry entry =
+ IDEMPOTENCY_STORE.compute(
+ key,
+ (k, current) -> {
+ if (current == null || current.isExpired()) {
+ isLeader.set(true);
+ return IdempotencyEntry.inProgress();
+ }
+ return current;
+ });
+
+ // Fast-path: already finalized (another request completed earlier)
+ if (entry.status == IdempotencyEntry.Status.FINALIZED) {
+ if (entry.error != null) {
+ throw entry.error;
+ }
+ return (T) entry.responseBody;
+ }
+
+ if (!isLeader.get()) {
+ // In-flight coalescing: wait for the leader request to finalize
+ entry.awaitFinalized();
+ if (entry.error != null) {
+ throw entry.error;
+ }
+ return (T) entry.responseBody;
+ }
+
+ // Leader request: execute the action and finalize the entry
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+ return res;
+ } catch (RuntimeException e) {
+ entry.finalizeError(e);
+ throw e;
+ }
+ }
+
+ @VisibleForTesting
+ static void setIdempotencyLifetimeFromIso(String isoDuration) {
+ if (isoDuration == null) {
+ return;
+ }
+ try {
+ idempotencyLifetimeMillis = Duration.parse(isoDuration).toMillis();
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Invalid idempotency lifetime: " + isoDuration, e);
+ }
+ }
+
+ private static final class IdempotencyEntry {
+ enum Status {
+ IN_PROGRESS,
+ FINALIZED
+ }
+
+ private final CountDownLatch latch;
+ private final long firstSeenMillis;
+ private volatile Status status;
+ private volatile Object responseBody;
+ private volatile RuntimeException error;
+
+ private IdempotencyEntry(Status status) {
+ this.status = status;
+ this.latch = new CountDownLatch(1);
+ this.firstSeenMillis = System.currentTimeMillis();
+ }
+
+ static IdempotencyEntry inProgress() {
+ return new IdempotencyEntry(Status.IN_PROGRESS);
+ }
+
+ void finalizeSuccess(Object body) {
+ this.responseBody = body;
+ this.status = Status.FINALIZED;
+ this.latch.countDown();
+ }
+
+ void finalizeError(RuntimeException cause) {
+ this.error = cause;
+ this.status = Status.FINALIZED;
+ this.latch.countDown();
+ }
+
+ void awaitFinalized() {
Review Comment:
maybe `awaitFinalization`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#issuecomment-3657055332
@nastra @amogh-jahagirdar @singhpk234 Could you please take a look when you have a moment? Thanks!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2667317336
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +114,147 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ AtomicBoolean isLeader = new AtomicBoolean(false);
Review Comment:
makes sense can we please add this as code comment, wdyt ?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2674261408
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
+ if (!expired) {
+ existing.awaitFinalized();
+ if (existing.error != null) {
+ throw existing.error;
+ }
+ return (T) existing.responseBody;
+ } else {
+ IDEMPOTENCY_STORE.remove(key, existing);
+ }
+ }
+
+ IdempotencyEntry entry = IdempotencyEntry.inProgress();
+ IDEMPOTENCY_STORE.put(key, entry);
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+
+ if (SIMULATE_503_ON_FIRST_SUCCESS_KEYS.remove(key)) {
+ throw new CommitStateUnknownException(
+ new RuntimeException("simulated transient 503 after success"));
+ }
+ return res;
+ } catch (RuntimeException e) {
+ if (entry.status != IdempotencyEntry.Status.FINALIZED || entry.responseBody == null) {
+ entry.finalizeError(e);
+ }
+ throw e;
+ }
+ }
+
+ // Test hooks/configuration for idempotency behavior
+ public static void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
Thanks — agreed. I removed the test hook from `CatalogHandlers` and kept R`ESTCatalogAdapter` free of test knobs.
The transient-503 simulation is now implemented only in a test-only adapter used by `TestRESTCatalog` (`HeaderValidatingAdapter`), so production-ish code paths don’t carry test conditions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2676150344
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -124,12 +133,107 @@ public class TestRESTCatalog extends CatalogTests<RESTCatalog> {
RESTCatalogProperties.NAMESPACE_SEPARATOR,
RESTCatalogAdapter.NAMESPACE_SEPARATOR_URLENCODED_UTF_8));
+ /**
+ * Test-only adapter that keeps request/response round-trip serialization and header validation
+ * from the base test setup, while also allowing specific tests to inject transient failures.
+ */
+ private static class HeaderValidatingAdapter extends RESTCatalogAdapter {
+ private final HTTPHeaders catalogHeaders;
+ private final HTTPHeaders contextHeaders;
+ private final Set<String> simulate503OnFirstSuccessKeys =
+ org.apache.iceberg.relocated.com.google.common.collect.Sets.newConcurrentHashSet();
Review Comment:
Minor: Maybe this should be a mapping from the key to a generic throwable that's thrown, then we could add other failure cases and make sure idempotency returns or throws as expected.
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +115,152 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // "Leader" is the request thread that wins the IDEMPOTENCY_STORE.compute(...) and creates (or
+ // replaces) the IN_PROGRESS entry for this Idempotency-Key. Only the leader executes the
+ // action and finalizes the entry; concurrent requests for the same key ("followers") wait on
+ // the latch and then replay the finalized result/error.
+ AtomicBoolean isLeader = new AtomicBoolean(false);
Review Comment:
Minor: I'd probably just call this isFirst. Leader/follower makes it seem more than it really is.
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3313,6 +3393,155 @@ public void testClientDoesNotSendIdempotencyWhenServerNotAdvertising() {
local.dropTable(ident);
}
+ @Test
+ public void testIdempotentDuplicateCreateReturnsCached() {
+ String key = "dup-create-key";
+ Namespace ns = Namespace.of("ns_dup");
+ Pair<TableIdentifier, Pair<RESTClient, Map<String, String>>> env =
+ prepareIdempotentEnv(key, ns, "t_dup");
+ TableIdentifier ident = env.first();
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = env.second();
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
Review Comment:
Nit: I think this bit is repeated across tests, any way to abstract some of this behind a helper?
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -268,19 +277,29 @@ public <T extends RESTResponse> T handleRequest(
return castResponse(
responseType, CatalogHandlers.stageTableCreate(catalog, namespace, request));
} else {
- LoadTableResponse response = CatalogHandlers.createTable(catalog, namespace, request);
- responseHeaders.accept(
- ImmutableMap.of(HttpHeaders.ETAG, ETagProvider.of(response.metadataLocation())));
- return castResponse(responseType, response);
+ return CatalogHandlers.withIdempotency(
+ httpRequest,
+ () -> {
+ LoadTableResponse response =
+ CatalogHandlers.createTable(catalog, namespace, request);
+ responseHeaders.accept(
+ ImmutableMap.of(
+ HttpHeaders.ETAG, ETagProvider.of(response.metadataLocation())));
+ return castResponse(responseType, response);
+ });
}
}
case DROP_TABLE:
{
if (PropertyUtil.propertyAsBoolean(vars, "purgeRequested", false)) {
- CatalogHandlers.purgeTable(catalog, tableIdentFromPathVars(vars));
+ CatalogHandlers.withIdempotency(
Review Comment:
Minor: I wonder if withIdempotency even needs to be exposed? Feels like it could just be invoked internally in each of the existing CatalogHandlers methods?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "geruh (via GitHub)" <gi...@apache.org>.
geruh commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2644099428
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,140 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
Review Comment:
This is rare in this use case but maybe we should use `computeIfAbsent` here to avoid any race conditions get and put issues
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2606789277
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
Review Comment:
you could also make this a function on the entry itself, so that you'd only call `if(!entry.expired())` here
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2608647370
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
+ if (!expired) {
+ existing.awaitFinalized();
+ if (existing.error != null) {
+ throw existing.error;
+ }
+ return (T) existing.responseBody;
+ } else {
+ IDEMPOTENCY_STORE.remove(key, existing);
+ }
+ }
+
+ IdempotencyEntry entry = IdempotencyEntry.inProgress();
+ IDEMPOTENCY_STORE.put(key, entry);
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+
+ if (SIMULATE_503_ON_FIRST_SUCCESS_KEYS.remove(key)) {
+ throw new CommitStateUnknownException(
+ new RuntimeException("simulated transient 503 after success"));
+ }
+ return res;
+ } catch (RuntimeException e) {
+ if (entry.status != IdempotencyEntry.Status.FINALIZED || entry.responseBody == null) {
+ entry.finalizeError(e);
+ }
+ throw e;
+ }
+ }
+
+ // Test hooks/configuration for idempotency behavior
+ public static void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
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(...).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2674257433
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +114,147 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ AtomicBoolean isLeader = new AtomicBoolean(false);
Review Comment:
Comment was added. Thanks
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2677315825
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -124,12 +133,107 @@ public class TestRESTCatalog extends CatalogTests<RESTCatalog> {
RESTCatalogProperties.NAMESPACE_SEPARATOR,
RESTCatalogAdapter.NAMESPACE_SEPARATOR_URLENCODED_UTF_8));
+ /**
+ * Test-only adapter that keeps request/response round-trip serialization and header validation
+ * from the base test setup, while also allowing specific tests to inject transient failures.
+ */
+ private static class HeaderValidatingAdapter extends RESTCatalogAdapter {
+ private final HTTPHeaders catalogHeaders;
+ private final HTTPHeaders contextHeaders;
+ private final Set<String> simulate503OnFirstSuccessKeys =
+ org.apache.iceberg.relocated.com.google.common.collect.Sets.newConcurrentHashSet();
Review Comment:
Good idea. I generalized the test-only hook to a key -> RuntimeException map so tests can simulate different post-success transient failures while still keeping simulate503OnFirstSuccessForKey as a convenience wrapper for the current case.
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -268,19 +277,29 @@ public <T extends RESTResponse> T handleRequest(
return castResponse(
responseType, CatalogHandlers.stageTableCreate(catalog, namespace, request));
} else {
- LoadTableResponse response = CatalogHandlers.createTable(catalog, namespace, request);
- responseHeaders.accept(
- ImmutableMap.of(HttpHeaders.ETAG, ETagProvider.of(response.metadataLocation())));
- return castResponse(responseType, response);
+ return CatalogHandlers.withIdempotency(
+ httpRequest,
+ () -> {
+ LoadTableResponse response =
+ CatalogHandlers.createTable(catalog, namespace, request);
+ responseHeaders.accept(
+ ImmutableMap.of(
+ HttpHeaders.ETAG, ETagProvider.of(response.metadataLocation())));
+ return castResponse(responseType, response);
+ });
}
}
case DROP_TABLE:
{
if (PropertyUtil.propertyAsBoolean(vars, "purgeRequested", false)) {
- CatalogHandlers.purgeTable(catalog, tableIdentFromPathVars(vars));
+ CatalogHandlers.withIdempotency(
Review Comment:
Good point. The idempotency behavior is driven by the HTTP request headers, so I’m keeping the wrapping at the routing/adapter layer rather than pushing HTTPRequest through every CatalogHandlers.* method.
That said, I agree it doesn’t need to be exposed publicly — I made withIdempotency(...) package-private and it’s only used internally by the REST adapter/routes.
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3313,6 +3393,155 @@ public void testClientDoesNotSendIdempotencyWhenServerNotAdvertising() {
local.dropTable(ident);
}
+ @Test
+ public void testIdempotentDuplicateCreateReturnsCached() {
+ String key = "dup-create-key";
+ Namespace ns = Namespace.of("ns_dup");
+ Pair<TableIdentifier, Pair<RESTClient, Map<String, String>>> env =
+ prepareIdempotentEnv(key, ns, "t_dup");
+ TableIdentifier ident = env.first();
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = env.second();
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
Review Comment:
I factored the repeated prepareIdempotentEnv unpacking into a small IdempotentEnv helper
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +115,152 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // "Leader" is the request thread that wins the IDEMPOTENCY_STORE.compute(...) and creates (or
+ // replaces) the IN_PROGRESS entry for this Idempotency-Key. Only the leader executes the
+ // action and finalizes the entry; concurrent requests for the same key ("followers") wait on
+ // the latch and then replay the finalized result/error.
+ AtomicBoolean isLeader = new AtomicBoolean(false);
Review Comment:
Changed to `isFirst`. Thanks
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#issuecomment-3726791453
@nastra I removed the test hook from CatalogHandlers and kept RESTCatalogAdapter free of test knobs.
The transient-503 simulation is now implemented only in a test-only adapter used by TestRESTCatalog (HeaderValidatingAdapter). Could you please check if this looks OK to you? Thanks!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2636621534
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -122,6 +124,11 @@ public RESTCatalogAdapter(Catalog catalog) {
this.asViewCatalog = catalog instanceof ViewCatalog ? (ViewCatalog) catalog : null;
}
+ /** Test helper to simulate a transient 503 after the first successful mutation for a key. */
+ public void simulate503OnFirstSuccessForKey(String key) {
+ simulate503OnFirstSuccessKeys.add(key);
+ }
+
private static OAuthTokenResponse handleOAuthRequest(Object body) {
Review Comment:
can we wrap this to plan-api endpoints too ?
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -204,8 +211,11 @@ public <T extends RESTResponse> T handleRequest(
case CREATE_NAMESPACE:
if (asNamespaceCatalog != null) {
CreateNamespaceRequest request = castRequest(CreateNamespaceRequest.class, body);
- return castResponse(
- responseType, CatalogHandlers.createNamespace(asNamespaceCatalog, request));
+ return CatalogHandlers.withIdempotency(
Review Comment:
can we add this for mutable remote scan plan endpoint ?
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -122,6 +124,11 @@ public RESTCatalogAdapter(Catalog catalog) {
this.asViewCatalog = catalog instanceof ViewCatalog ? (ViewCatalog) catalog : null;
}
+ /** Test helper to simulate a transient 503 after the first successful mutation for a key. */
+ public void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
nit : should we make it generic like simulate status code or something ?
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3312,6 +3319,160 @@ public void testClientDoesNotSendIdempotencyWhenServerNotAdvertising() {
local.dropTable(ident);
}
+ @Test
+ public void testIdempotentDuplicateCreateReturnsCached() {
+ String key = "dup-create-key";
+ Namespace ns = Namespace.of("ns_dup");
+ TableIdentifier ident = TableIdentifier.of(ns, "t_dup");
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = httpAndHeaders(key);
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First create succeeds
+ LoadTableResponse first =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(first).isNotNull();
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // Duplicate with same key returns cached 200 OK
+ LoadTableResponse second =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(second).isNotNull();
+
+ // Clean up
+ restCatalog.dropTable(ident);
+ }
+
+ @Test
+ public void testIdempotencyKeyLifetimeExpiredTreatsAsNew() {
+ // Set TTL to 0 so cached success expires immediately
+ CatalogHandlers.setIdempotencyLifetimeFromIso("PT0S");
+ try {
+ String key = "expired-create-key";
+ Namespace ns = Namespace.of("ns_exp");
+ TableIdentifier ident = TableIdentifier.of(ns, "t_exp");
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = httpAndHeaders(key);
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First create succeeds
+ LoadTableResponse created =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(created).isNotNull();
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // TTL expired -> duplicate with same key should be treated as new and fail with AlreadyExists
+ assertThatThrownBy(
+ () ->
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler()))
+ .isInstanceOf(AlreadyExistsException.class)
+ .hasMessageContaining(ident.toString());
+
+ // Clean up
+ restCatalog.dropTable(ident);
+ } finally {
+ // Restore default TTL for other tests
+ CatalogHandlers.setIdempotencyLifetimeFromIso("PT30M");
+ }
+ }
+
+ @Test
+ public void testIdempotentCreateReplayAfterSimulated503() {
+ // Use a fixed key and simulate 503 after first success for that key
+ String key = "idemp-create-503";
+ adapterForRESTServer.simulate503OnFirstSuccessForKey(key);
+ Namespace ns = Namespace.of("ns_idemp");
+ TableIdentifier ident = TableIdentifier.of(ns, "t_idemp");
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = httpAndHeaders(key);
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First attempt: server finalizes success but responds 503
+ assertThatThrownBy(
+ () ->
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler()))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("simulated transient 503");
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // Retry with same key: server should replay 200 OK
+ LoadTableResponse replay =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(replay).isNotNull();
+
+ // Clean up
+ restCatalog.dropTable(ident);
+ }
+
+ @Test
+ public void testIdempotentDropDuplicateNoop() {
+ 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();
Review Comment:
i wonder if all this can be abstracted in a utils which takes key, namespace and table_name ? wdyt ?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2663269683
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3312,6 +3319,164 @@ public void testClientDoesNotSendIdempotencyWhenServerNotAdvertising() {
local.dropTable(ident);
}
+ @Test
+ public void testIdempotentDuplicateCreateReturnsCached() {
+ String key = "dup-create-key";
+ Namespace ns = Namespace.of("ns_dup");
+ Pair<TableIdentifier, Pair<RESTClient, Map<String, String>>> env =
+ prepareIdempotentEnv(key, ns, "t_dup");
+ TableIdentifier ident = env.first();
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = env.second();
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First create succeeds
+ LoadTableResponse first =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(first).isNotNull();
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // Duplicate with same key returns cached 200 OK
+ LoadTableResponse second =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(second).isNotNull();
+
+ // Clean up
+ restCatalog.dropTable(ident);
+ }
+
+ @Test
+ public void testIdempotencyKeyLifetimeExpiredTreatsAsNew() {
+ // Set TTL to 0 so cached success expires immediately
+ CatalogHandlers.setIdempotencyLifetimeFromIso("PT0S");
+ try {
+ String key = "expired-create-key";
+ Namespace ns = Namespace.of("ns_exp");
+ Pair<TableIdentifier, Pair<RESTClient, Map<String, String>>> env =
+ prepareIdempotentEnv(key, ns, "t_exp");
+ TableIdentifier ident = env.first();
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = env.second();
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First create succeeds
+ LoadTableResponse created =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(created).isNotNull();
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // TTL expired -> duplicate with same key should be treated as new and fail with AlreadyExists
+ assertThatThrownBy(
+ () ->
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler()))
+ .isInstanceOf(AlreadyExistsException.class)
+ .hasMessageContaining(ident.toString());
+
+ // Clean up
+ restCatalog.dropTable(ident);
Review Comment:
do we need explicit drop ?
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +114,147 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ AtomicBoolean isLeader = new AtomicBoolean(false);
+ IdempotencyEntry entry =
+ IDEMPOTENCY_STORE.compute(
+ key,
+ (k, current) -> {
+ if (current == null || current.isExpired()) {
+ isLeader.set(true);
+ return IdempotencyEntry.inProgress();
+ }
+ return current;
+ });
+
+ // Fast-path: already finalized (another request completed earlier)
+ if (entry.status == IdempotencyEntry.Status.FINALIZED) {
+ if (entry.error != null) {
+ throw entry.error;
+ }
+ return (T) entry.responseBody;
+ }
+
+ if (!isLeader.get()) {
+ // In-flight coalescing: wait for the leader request to finalize
+ entry.awaitFinalized();
+ if (entry.error != null) {
+ throw entry.error;
+ }
+ return (T) entry.responseBody;
+ }
+
+ // Leader request: execute the action and finalize the entry
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+ return res;
+ } catch (RuntimeException e) {
+ entry.finalizeError(e);
+ throw e;
+ }
+ }
+
+ public static void setIdempotencyLifetimeFromIso(String isoDuration) {
+ if (isoDuration == null) {
+ return;
+ }
+ try {
+ idempotencyLifetimeMillis = Duration.parse(isoDuration).toMillis();
+ } catch (Exception e) {
+ // ignore parse errors; keep default
Review Comment:
sorry i might have missed this in first pass, this is public imho we should throw illegalArgs ?
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +114,147 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ AtomicBoolean isLeader = new AtomicBoolean(false);
Review Comment:
whats a Leader ?
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -114,6 +115,7 @@ public class RESTCatalogAdapter extends BaseHTTPClient {
private AuthSession authSession = AuthSession.EMPTY;
private PlanningBehavior planningBehavior;
+ private final java.util.Set<String> simulate503OnFirstSuccessKeys = Sets.newConcurrentHashSet();
Review Comment:
nit: can we import this instead of inline ?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2668698620
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
+ if (!expired) {
+ existing.awaitFinalized();
+ if (existing.error != null) {
+ throw existing.error;
+ }
+ return (T) existing.responseBody;
+ } else {
+ IDEMPOTENCY_STORE.remove(key, existing);
+ }
+ }
+
+ IdempotencyEntry entry = IdempotencyEntry.inProgress();
+ IDEMPOTENCY_STORE.put(key, entry);
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+
+ if (SIMULATE_503_ON_FIRST_SUCCESS_KEYS.remove(key)) {
+ throw new CommitStateUnknownException(
+ new RuntimeException("simulated transient 503 after success"));
+ }
+ return res;
+ } catch (RuntimeException e) {
+ if (entry.status != IdempotencyEntry.Status.FINALIZED || entry.responseBody == null) {
+ entry.finalizeError(e);
+ }
+ throw e;
+ }
+ }
+
+ // Test hooks/configuration for idempotency behavior
+ public static void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
I don't think we should add this functionality at all to the adapter. Instead, what about just having a custom adapter that simulates this behavior for that particular test case. We already use custom adapters with custom behavior in a lot of other test methods.
My main concern is that we're trying to force "test conditions" into code that shouldn't have such test conditions, because `RESTCatalogAdapter` could be used in real usage scenarios and it shouldn't ideally contain test code. Additionally, this opens up a precedent for future code modifications where it might seem ok to add more and more testing conditions to the adapter or to CatalogHandlers
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2674261408
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
+ if (!expired) {
+ existing.awaitFinalized();
+ if (existing.error != null) {
+ throw existing.error;
+ }
+ return (T) existing.responseBody;
+ } else {
+ IDEMPOTENCY_STORE.remove(key, existing);
+ }
+ }
+
+ IdempotencyEntry entry = IdempotencyEntry.inProgress();
+ IDEMPOTENCY_STORE.put(key, entry);
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+
+ if (SIMULATE_503_ON_FIRST_SUCCESS_KEYS.remove(key)) {
+ throw new CommitStateUnknownException(
+ new RuntimeException("simulated transient 503 after success"));
+ }
+ return res;
+ } catch (RuntimeException e) {
+ if (entry.status != IdempotencyEntry.Status.FINALIZED || entry.responseBody == null) {
+ entry.finalizeError(e);
+ }
+ throw e;
+ }
+ }
+
+ // Test hooks/configuration for idempotency behavior
+ public static void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
Thanks — agreed. I removed the test hook from `CatalogHandlers` and kept R`ESTCatalogAdapter` free of test knobs.
The transient-503 simulation is now implemented only in a test-only adapter used by T`estRESTCatalog` (`HeaderValidatingAdapter`), so production-ish code paths don’t carry test conditions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao merged PR #14773:
URL: https://github.com/apache/iceberg/pull/14773
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2609419993
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -204,8 +211,11 @@ public <T extends RESTResponse> T handleRequest(
case CREATE_NAMESPACE:
if (asNamespaceCatalog != null) {
CreateNamespaceRequest request = castRequest(CreateNamespaceRequest.class, body);
- return castResponse(
- responseType, CatalogHandlers.createNamespace(asNamespaceCatalog, request));
+ return CatalogHandlers.withIdempotency(
Review Comment:
can we add this for mutable remote scan plan endpoint ?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2663686041
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +114,147 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ AtomicBoolean isLeader = new AtomicBoolean(false);
+ IdempotencyEntry entry =
+ IDEMPOTENCY_STORE.compute(
+ key,
+ (k, current) -> {
+ if (current == null || current.isExpired()) {
+ isLeader.set(true);
+ return IdempotencyEntry.inProgress();
+ }
+ return current;
+ });
+
+ // Fast-path: already finalized (another request completed earlier)
+ if (entry.status == IdempotencyEntry.Status.FINALIZED) {
+ if (entry.error != null) {
+ throw entry.error;
+ }
+ return (T) entry.responseBody;
+ }
+
+ if (!isLeader.get()) {
+ // In-flight coalescing: wait for the leader request to finalize
+ entry.awaitFinalized();
+ if (entry.error != null) {
+ throw entry.error;
+ }
+ return (T) entry.responseBody;
+ }
+
+ // Leader request: execute the action and finalize the entry
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+ return res;
+ } catch (RuntimeException e) {
+ entry.finalizeError(e);
+ throw e;
+ }
+ }
+
+ public static void setIdempotencyLifetimeFromIso(String isoDuration) {
+ if (isoDuration == null) {
+ return;
+ }
+ try {
+ idempotencyLifetimeMillis = Duration.parse(isoDuration).toMillis();
+ } catch (Exception e) {
+ // ignore parse errors; keep default
Review Comment:
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.
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -114,6 +115,7 @@ public class RESTCatalogAdapter extends BaseHTTPClient {
private AuthSession authSession = AuthSession.EMPTY;
private PlanningBehavior planningBehavior;
+ private final java.util.Set<String> simulate503OnFirstSuccessKeys = Sets.newConcurrentHashSet();
Review Comment:
Fixed. Thanks
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2668719311
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -122,6 +124,11 @@ public RESTCatalogAdapter(Catalog catalog) {
this.asViewCatalog = catalog instanceof ViewCatalog ? (ViewCatalog) catalog : null;
}
+ /** Test helper to simulate a transient 503 after the first successful mutation for a key. */
+ public void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
It would be great to not add test-specific code to the adapter/catalog handler. See my other comment on this: https://github.com/apache/iceberg/pull/14773#discussion_r2668698620
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2669159827
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
+ if (!expired) {
+ existing.awaitFinalized();
+ if (existing.error != null) {
+ throw existing.error;
+ }
+ return (T) existing.responseBody;
+ } else {
+ IDEMPOTENCY_STORE.remove(key, existing);
+ }
+ }
+
+ IdempotencyEntry entry = IdempotencyEntry.inProgress();
+ IDEMPOTENCY_STORE.put(key, entry);
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+
+ if (SIMULATE_503_ON_FIRST_SUCCESS_KEYS.remove(key)) {
+ throw new CommitStateUnknownException(
+ new RuntimeException("simulated transient 503 after success"));
+ }
+ return res;
+ } catch (RuntimeException e) {
+ if (entry.status != IdempotencyEntry.Status.FINALIZED || entry.responseBody == null) {
+ entry.finalizeError(e);
+ }
+ throw e;
+ }
+ }
+
+ // Test hooks/configuration for idempotency behavior
+ public static void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
I guess one alternative would be to have a custom adapter in the testFixtures, similar to https://github.com/apache/iceberg/blob/9ea3b136c2d06af1d5167900b7253e7b850db9f4/open-api/src/testFixtures/java/org/apache/iceberg/rest/RESTServerCatalogAdapter.java#L30 where we have a custom adapter that performs credential vending
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2606798937
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,145 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static final Set<String> SIMULATE_503_ON_FIRST_SUCCESS_KEYS = Sets.newConcurrentHashSet();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
+ if (existing != null) {
+ long now = System.currentTimeMillis();
+ boolean expired =
+ existing.status == IdempotencyEntry.Status.FINALIZED
+ && (now - existing.firstSeenMillis) > idempotencyLifetimeMillis;
+ if (!expired) {
+ existing.awaitFinalized();
+ if (existing.error != null) {
+ throw existing.error;
+ }
+ return (T) existing.responseBody;
+ } else {
+ IDEMPOTENCY_STORE.remove(key, existing);
+ }
+ }
+
+ IdempotencyEntry entry = IdempotencyEntry.inProgress();
+ IDEMPOTENCY_STORE.put(key, entry);
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+
+ if (SIMULATE_503_ON_FIRST_SUCCESS_KEYS.remove(key)) {
+ throw new CommitStateUnknownException(
+ new RuntimeException("simulated transient 503 after success"));
+ }
+ return res;
+ } catch (RuntimeException e) {
+ if (entry.status != IdempotencyEntry.Status.FINALIZED || entry.responseBody == null) {
+ entry.finalizeError(e);
+ }
+ throw e;
+ }
+ }
+
+ // Test hooks/configuration for idempotency behavior
+ public static void simulate503OnFirstSuccessForKey(String key) {
Review Comment:
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
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#issuecomment-3761158311
I merged this PR. If there are any more comments, I will address in follow-ups. Thank you all for reviewing this PR!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2603651256
##########
core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java:
##########
@@ -557,6 +568,19 @@ protected <T extends RESTResponse> T execute(
vars.putAll(request.queryParameters());
vars.putAll(routeAndVars.second());
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ request.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ boolean isMutation =
+ request.method() == HTTPMethod.POST || request.method() == HTTPMethod.DELETE;
+ if (isMutation && keyHeader.isPresent()) {
+ T response =
+ handleIdempotentMutation(
Review Comment:
Fixed. Could you please take a look again?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2658490590
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3312,6 +3319,160 @@ public void testClientDoesNotSendIdempotencyWhenServerNotAdvertising() {
local.dropTable(ident);
}
+ @Test
+ public void testIdempotentDuplicateCreateReturnsCached() {
+ String key = "dup-create-key";
+ Namespace ns = Namespace.of("ns_dup");
+ TableIdentifier ident = TableIdentifier.of(ns, "t_dup");
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = httpAndHeaders(key);
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First create succeeds
+ LoadTableResponse first =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(first).isNotNull();
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // Duplicate with same key returns cached 200 OK
+ LoadTableResponse second =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(second).isNotNull();
+
+ // Clean up
+ restCatalog.dropTable(ident);
+ }
+
+ @Test
+ public void testIdempotencyKeyLifetimeExpiredTreatsAsNew() {
+ // Set TTL to 0 so cached success expires immediately
+ CatalogHandlers.setIdempotencyLifetimeFromIso("PT0S");
+ try {
+ String key = "expired-create-key";
+ Namespace ns = Namespace.of("ns_exp");
+ TableIdentifier ident = TableIdentifier.of(ns, "t_exp");
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = httpAndHeaders(key);
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First create succeeds
+ LoadTableResponse created =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(created).isNotNull();
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // TTL expired -> duplicate with same key should be treated as new and fail with AlreadyExists
+ assertThatThrownBy(
+ () ->
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler()))
+ .isInstanceOf(AlreadyExistsException.class)
+ .hasMessageContaining(ident.toString());
+
+ // Clean up
+ restCatalog.dropTable(ident);
+ } finally {
+ // Restore default TTL for other tests
+ CatalogHandlers.setIdempotencyLifetimeFromIso("PT30M");
+ }
+ }
+
+ @Test
+ public void testIdempotentCreateReplayAfterSimulated503() {
+ // Use a fixed key and simulate 503 after first success for that key
+ String key = "idemp-create-503";
+ adapterForRESTServer.simulate503OnFirstSuccessForKey(key);
+ Namespace ns = Namespace.of("ns_idemp");
+ TableIdentifier ident = TableIdentifier.of(ns, "t_idemp");
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = httpAndHeaders(key);
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First attempt: server finalizes success but responds 503
+ assertThatThrownBy(
+ () ->
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler()))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("simulated transient 503");
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // Retry with same key: server should replay 200 OK
+ LoadTableResponse replay =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(replay).isNotNull();
+
+ // Clean up
+ restCatalog.dropTable(ident);
+ }
+
+ @Test
+ public void testIdempotentDropDuplicateNoop() {
+ 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();
Review Comment:
Thanks for the comment! I have added a smaller helper method for the common code.
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -105,9 +111,140 @@ public class CatalogHandlers {
private static final InMemoryPlanningState IN_MEMORY_PLANNING_STATE =
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
private CatalogHandlers() {}
+ /**
+ * Execute a mutation with basic idempotency semantics based on the Idempotency-Key header.
+ *
+ * <p>This simple reference implementation stores the response in-memory keyed by the header
+ * value. If the same key is seen again, the stored response is returned and the action is not
+ * re-executed. This is suitable for tests and examples; production servers should provide a
+ * durable store and TTL management.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ // check existing entry and TTL
+ IdempotencyEntry existing = IDEMPOTENCY_STORE.get(key);
Review Comment:
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.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2663685626
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +114,147 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ AtomicBoolean isLeader = new AtomicBoolean(false);
Review Comment:
“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.
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3312,6 +3319,164 @@ public void testClientDoesNotSendIdempotencyWhenServerNotAdvertising() {
local.dropTable(ident);
}
+ @Test
+ public void testIdempotentDuplicateCreateReturnsCached() {
+ String key = "dup-create-key";
+ Namespace ns = Namespace.of("ns_dup");
+ Pair<TableIdentifier, Pair<RESTClient, Map<String, String>>> env =
+ prepareIdempotentEnv(key, ns, "t_dup");
+ TableIdentifier ident = env.first();
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = env.second();
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First create succeeds
+ LoadTableResponse first =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(first).isNotNull();
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // Duplicate with same key returns cached 200 OK
+ LoadTableResponse second =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(second).isNotNull();
+
+ // Clean up
+ restCatalog.dropTable(ident);
+ }
+
+ @Test
+ public void testIdempotencyKeyLifetimeExpiredTreatsAsNew() {
+ // Set TTL to 0 so cached success expires immediately
+ CatalogHandlers.setIdempotencyLifetimeFromIso("PT0S");
+ try {
+ String key = "expired-create-key";
+ Namespace ns = Namespace.of("ns_exp");
+ Pair<TableIdentifier, Pair<RESTClient, Map<String, String>>> env =
+ prepareIdempotentEnv(key, ns, "t_exp");
+ TableIdentifier ident = env.first();
+ Pair<RESTClient, Map<String, String>> httpAndHeaders = env.second();
+ RESTClient http = httpAndHeaders.first();
+ Map<String, String> headers = httpAndHeaders.second();
+ CreateTableRequest req = createReq(ident);
+
+ // First create succeeds
+ LoadTableResponse created =
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler());
+ assertThat(created).isNotNull();
+
+ // Verify request shape (method, path, headers including Idempotency-Key)
+ verifyCreatePost(ns, headers);
+
+ // TTL expired -> duplicate with same key should be treated as new and fail with AlreadyExists
+ assertThatThrownBy(
+ () ->
+ http.post(
+ ResourcePaths.forCatalogProperties(ImmutableMap.of()).tables(ns),
+ req,
+ LoadTableResponse.class,
+ headers,
+ ErrorHandlers.tableErrorHandler()))
+ .isInstanceOf(AlreadyExistsException.class)
+ .hasMessageContaining(ident.toString());
+
+ // Clean up
+ restCatalog.dropTable(ident);
Review Comment:
Good catch. This is not needed. Removed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2608647979
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3345,6 +3470,73 @@ public <T extends RESTResponse> T execute(
return local;
}
+ private IdempotentCreateEnv prepareIdempotentCreateEnv(
+ String namespaceName, String tableName, String idempotencyKey) {
+ Namespace ns = Namespace.of(namespaceName);
+ TableIdentifier ident = TableIdentifier.of(ns, tableName);
+ Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
+
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+
+ Map<String, String> headers =
+ ImmutableMap.of(
+ RESTUtil.IDEMPOTENCY_KEY_HEADER,
+ idempotencyKey,
+ "Authorization",
+ "Bearer client-credentials-token:sub=user",
+ "test-header",
+ "test-value");
+
+ Map<String, String> conf =
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ httpServer.getURI().toString(),
+ HTTPClient.REST_SOCKET_TIMEOUT_MS,
+ "600000",
+ HTTPClient.REST_CONNECTION_TIMEOUT_MS,
+ "600000",
+ "header.test-header",
+ "test-value");
+ RESTClient httpBase =
+ HTTPClient.builder(conf)
+ .uri(conf.get(CatalogProperties.URI))
+ .withHeaders(RESTUtil.configHeaders(conf))
+ .build();
+ AuthManager am = AuthManagers.loadAuthManager("test", conf);
+ AuthSession httpSession = am.initSession(httpBase, conf);
+ RESTClient http = httpBase.withAuthSession(httpSession);
+
+ CreateTableRequest req =
Review Comment:
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.
##########
core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java:
##########
@@ -3345,6 +3470,73 @@ public <T extends RESTResponse> T execute(
return local;
}
+ private IdempotentCreateEnv prepareIdempotentCreateEnv(
+ String namespaceName, String tableName, String idempotencyKey) {
+ Namespace ns = Namespace.of(namespaceName);
+ TableIdentifier ident = TableIdentifier.of(ns, tableName);
+ Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
+
+ restCatalog.createNamespace(ns, ImmutableMap.of());
+
+ Map<String, String> headers =
+ ImmutableMap.of(
+ RESTUtil.IDEMPOTENCY_KEY_HEADER,
+ idempotencyKey,
+ "Authorization",
+ "Bearer client-credentials-token:sub=user",
+ "test-header",
+ "test-value");
+
+ Map<String, String> conf =
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ httpServer.getURI().toString(),
+ HTTPClient.REST_SOCKET_TIMEOUT_MS,
+ "600000",
+ HTTPClient.REST_CONNECTION_TIMEOUT_MS,
+ "600000",
+ "header.test-header",
+ "test-value");
+ RESTClient httpBase =
+ HTTPClient.builder(conf)
+ .uri(conf.get(CatalogProperties.URI))
+ .withHeaders(RESTUtil.configHeaders(conf))
+ .build();
+ AuthManager am = AuthManagers.loadAuthManager("test", conf);
+ AuthSession httpSession = am.initSession(httpBase, conf);
+ RESTClient http = httpBase.withAuthSession(httpSession);
+
+ CreateTableRequest req =
+ CreateTableRequest.builder()
+ .withName(ident.name())
+ .withSchema(schema)
+ .withPartitionSpec(PartitionSpec.unpartitioned())
+ .build();
+
+ return new IdempotentCreateEnv(ns, ident, headers, http, req);
+ }
+
+ private static class IdempotentCreateEnv {
Review Comment:
I removed this wrapped class
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add idempotency adapter and E2E coverage [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #14773:
URL: https://github.com/apache/iceberg/pull/14773#discussion_r2674256626
##########
core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java:
##########
@@ -108,8 +115,148 @@ public class CatalogHandlers {
InMemoryPlanningState.getInstance();
private static final ExecutorService ASYNC_PLANNING_POOL = Executors.newSingleThreadExecutor();
+ // Advanced idempotency store with TTL and in-flight coalescing.
+ //
+ // Note: This is a simple in-memory implementation meant for tests and lightweight usage.
+ // Production servers should provide a durable store.
+ private static final ConcurrentMap<String, IdempotencyEntry> IDEMPOTENCY_STORE =
+ Maps.newConcurrentMap();
+ private static volatile long idempotencyLifetimeMillis = TimeUnit.MINUTES.toMillis(30);
+
private CatalogHandlers() {}
+ @SuppressWarnings("unchecked")
+ public static <T extends RESTResponse> T withIdempotency(
+ HTTPRequest httpRequest, Supplier<T> action) {
+ return withIdempotencyInternal(httpRequest, action);
+ }
+
+ public static void withIdempotency(HTTPRequest httpRequest, Runnable action) {
+ withIdempotencyInternal(
+ httpRequest,
+ () -> {
+ action.run();
+ return Boolean.TRUE;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T withIdempotencyInternal(HTTPRequest httpRequest, Supplier<T> action) {
+ Optional<HTTPHeaders.HTTPHeader> keyHeader =
+ httpRequest.headers().firstEntry(RESTUtil.IDEMPOTENCY_KEY_HEADER);
+ if (keyHeader.isEmpty()) {
+ return action.get();
+ }
+
+ String key = keyHeader.get().value();
+
+ AtomicBoolean isLeader = new AtomicBoolean(false);
+ IdempotencyEntry entry =
+ IDEMPOTENCY_STORE.compute(
+ key,
+ (k, current) -> {
+ if (current == null || current.isExpired()) {
+ isLeader.set(true);
+ return IdempotencyEntry.inProgress();
+ }
+ return current;
+ });
+
+ // Fast-path: already finalized (another request completed earlier)
+ if (entry.status == IdempotencyEntry.Status.FINALIZED) {
+ if (entry.error != null) {
+ throw entry.error;
+ }
+ return (T) entry.responseBody;
+ }
+
+ if (!isLeader.get()) {
+ // In-flight coalescing: wait for the leader request to finalize
+ entry.awaitFinalized();
+ if (entry.error != null) {
+ throw entry.error;
+ }
+ return (T) entry.responseBody;
+ }
+
+ // Leader request: execute the action and finalize the entry
+ try {
+ T res = action.get();
+ entry.finalizeSuccess(res);
+ return res;
+ } catch (RuntimeException e) {
+ entry.finalizeError(e);
+ throw e;
+ }
+ }
+
+ @VisibleForTesting
+ static void setIdempotencyLifetimeFromIso(String isoDuration) {
+ if (isoDuration == null) {
+ return;
+ }
+ try {
+ idempotencyLifetimeMillis = Duration.parse(isoDuration).toMillis();
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Invalid idempotency lifetime: " + isoDuration, e);
+ }
+ }
+
+ private static final class IdempotencyEntry {
+ enum Status {
+ IN_PROGRESS,
+ FINALIZED
+ }
+
+ private final CountDownLatch latch;
+ private final long firstSeenMillis;
+ private volatile Status status;
+ private volatile Object responseBody;
+ private volatile RuntimeException error;
+
+ private IdempotencyEntry(Status status) {
+ this.status = status;
+ this.latch = new CountDownLatch(1);
+ this.firstSeenMillis = System.currentTimeMillis();
+ }
+
+ static IdempotencyEntry inProgress() {
+ return new IdempotencyEntry(Status.IN_PROGRESS);
+ }
+
+ void finalizeSuccess(Object body) {
+ this.responseBody = body;
+ this.status = Status.FINALIZED;
+ this.latch.countDown();
+ }
+
+ void finalizeError(RuntimeException cause) {
+ this.error = cause;
+ this.status = Status.FINALIZED;
+ this.latch.countDown();
+ }
+
+ void awaitFinalized() {
Review Comment:
Fixed. Thanks
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org