From 3cc42d22bcc40af14f71331fba7a97123d6ac8c6 Mon Sep 17 00:00:00 2001 From: vignesh_A Date: Fri, 17 Jul 2026 10:34:46 +0530 Subject: [PATCH 1/5] feat(events): add webhook event listener with HMAC signing and retry Add a generic webhook event listener (polaris.event-listener.types=webhook) that delivers Polaris events as JSON HTTP POST requests to a configurable endpoint, e.g. a SIEM or audit collector. - Payloads are signed with HMAC-SHA256 when polaris.event-listener.webhook.secret is set and sent as the X-Polaris-Signature-256 header, so receivers can verify authenticity. - Failed deliveries (non-2xx or connection errors) are retried with exponential backoff (max-attempts / retry-backoff) before being dropped and logged. - The listener receives sanitized events only (no RawEventAccess), so denylisted attributes such as credentials are never delivered. --- CHANGELOG.md | 1 + bom/build.gradle.kts | 1 + extensions/events/webhook/build.gradle.kts | 50 ++++ .../events/webhook/WebhookEventListener.java | 220 ++++++++++++++++ .../WebhookEventListenerConfiguration.java | 132 ++++++++++ .../webhook/WebhookEventListenerTest.java | 245 ++++++++++++++++++ gradle/projects.main.properties | 1 + .../src/main/resources/application.properties | 9 + runtime/server/build.gradle.kts | 1 + ...smallrye-polaris_event_listener_webhook.md | 37 +++ .../configuration/configuration-reference.md | 30 +++ tools/config-docs/site/build.gradle.kts | 1 + 12 files changed, 728 insertions(+) create mode 100644 extensions/events/webhook/build.gradle.kts create mode 100644 extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java create mode 100644 extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java create mode 100644 extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java create mode 100644 site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md diff --git a/CHANGELOG.md b/CHANGELOG.md index ec7a853939..a1dd037d56 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti - Added optional `sessionPolicy` field to `SigV4AuthenticationParameters` for catalog federation. When set, the IAM session policy JSON is attached to the STS AssumeRole request, allowing administrators to restrict vended credentials to only the required AWS services and actions (Principle of Least Privilege). - Python CLI: added `--scheme` to specify URL scheme - Added opt-in idempotency for `createTable` and `updateTable` in the Iceberg REST catalog. When enabled via `polaris.idempotency.enabled=true` (default `false`), a client-supplied `Idempotency-Key` header is embedded into the table entity and committed in the same transaction as the operation; a retry carrying the same key within the TTL window (`polaris.idempotency.ttl`, default `PT5M`) replays the original success instead of failing — with `AlreadyExists` for `createTable`, or with `CommitFailedException` for `updateTable` when the request's requirements no longer match the already-advanced table. +- Added a `webhook` event listener (`polaris.event-listener.types=webhook`) that delivers Polaris events as JSON HTTP POST requests to a configurable endpoint, for example a SIEM or audit collector. Deliveries are retried with exponential backoff (`polaris.event-listener.webhook.max-attempts`, `polaris.event-listener.webhook.retry-backoff`) on a best-effort basis, payloads can be signed with HMAC-SHA256 via `polaris.event-listener.webhook.secret` (sent as the `X-Polaris-Signature-256` header) so receivers can verify authenticity, and `polaris.event-listener.webhook.headers` allows custom headers such as `Authorization` for receivers that require them. ### Changes - The admin tool's `bootstrap` command is now idempotent: bootstrapping a realm that is already diff --git a/bom/build.gradle.kts b/bom/build.gradle.kts index 07dd8a0fdb..f23900462c 100644 --- a/bom/build.gradle.kts +++ b/bom/build.gradle.kts @@ -99,6 +99,7 @@ dependencies { api(project(":polaris-extensions-auth-opa")) api(project(":polaris-extensions-auth-ranger")) + api(project(":polaris-extensions-events-webhook")) api(project(":polaris-extensions-federation-bigquery")) api(project(":polaris-extensions-federation-hadoop")) api(project(":polaris-extensions-federation-hive")) diff --git a/extensions/events/webhook/build.gradle.kts b/extensions/events/webhook/build.gradle.kts new file mode 100644 index 0000000000..b1edb5b15d --- /dev/null +++ b/extensions/events/webhook/build.gradle.kts @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +plugins { + id("polaris-server") + id("org.kordamp.gradle.jandex") +} + +dependencies { + implementation(project(":polaris-core")) + implementation(project(":polaris-runtime-service")) + + implementation(platform(libs.iceberg.bom)) + implementation("org.apache.iceberg:iceberg-api") + + implementation(platform(libs.jackson.bom)) + implementation("com.fasterxml.jackson.core:jackson-databind") + + implementation(platform(libs.quarkus.bom)) + implementation("io.quarkus:quarkus-core") + implementation("io.quarkus:quarkus-jackson") + + compileOnly(libs.jakarta.enterprise.cdi.api) + compileOnly(libs.jakarta.inject.api) + compileOnly(libs.jakarta.annotation.api) + compileOnly(libs.smallrye.config.core) + + implementation(libs.slf4j.api) + implementation(libs.smallrye.common.annotation) + + testImplementation(libs.assertj.core) + testImplementation(libs.awaitility) + testImplementation(libs.mockito.junit.jupiter) +} diff --git a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java new file mode 100644 index 0000000000..1b9899db5a --- /dev/null +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java @@ -0,0 +1,220 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.extensions.events.webhook; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.smallrye.common.annotation.Identifier; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.security.InvalidKeyException; +import java.security.NoSuchAlgorithmException; +import java.time.Clock; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Delivers Polaris events to an HTTP(S) endpoint as JSON POST requests, optionally signed with + * HMAC-SHA256. Failed deliveries (non-2xx responses or connection errors) are retried with + * exponential backoff; events that exhaust all attempts are dropped and logged. + */ +@ApplicationScoped +@Identifier("webhook") +public class WebhookEventListener implements PolarisEventListener { + private static final Logger LOGGER = LoggerFactory.getLogger(WebhookEventListener.class); + + static final String SIGNATURE_HEADER = "X-Polaris-Signature-256"; + static final String EVENT_TYPE_HEADER = "X-Polaris-Event"; + + private final URI endpoint; + private final String secret; + private final Map headers; + private final Duration timeout; + private final int maxAttempts; + private final Duration retryBackoff; + private final ObjectMapper objectMapper; + private final Clock clock; + + private HttpClient client; + private ScheduledExecutorService retryExecutor; + + @Inject + public WebhookEventListener( + WebhookEventListenerConfiguration config, ObjectMapper objectMapper, Clock clock) { + this.endpoint = config.endpoint().orElse(null); + this.secret = config.secret().orElse(null); + this.headers = config.headers(); + this.timeout = config.timeout(); + this.maxAttempts = config.maxAttempts(); + this.retryBackoff = config.retryBackoff(); + this.objectMapper = objectMapper; + this.clock = clock; + } + + @PostConstruct + void start() { + if (endpoint == null) { + throw new IllegalStateException( + "The webhook event listener is enabled but polaris.event-listener.webhook.endpoint is" + + " not set"); + } + this.client = createHttpClient(); + this.retryExecutor = + Executors.newSingleThreadScheduledExecutor( + runnable -> { + Thread thread = new Thread(runnable, "polaris-webhook-event-listener-retry"); + thread.setDaemon(true); + return thread; + }); + } + + protected HttpClient createHttpClient() { + return HttpClient.newBuilder().connectTimeout(timeout).build(); + } + + @PreDestroy + void shutdown() { + if (retryExecutor != null) { + retryExecutor.shutdownNow(); + retryExecutor = null; + } + } + + @Override + public void onEvent(PolarisEvent event) { + String payload = toJson(event); + if (payload == null) { + return; + } + send(payload, event.type().name(), 1); + } + + private void send(String payload, String eventType, int attempt) { + HttpRequest.Builder requestBuilder = + HttpRequest.newBuilder(endpoint) + .timeout(timeout) + .header("Content-Type", "application/json") + .header(EVENT_TYPE_HEADER, eventType); + headers.forEach(requestBuilder::header); + requestBuilder.POST(HttpRequest.BodyPublishers.ofString(payload, StandardCharsets.UTF_8)); + if (secret != null) { + requestBuilder.header(SIGNATURE_HEADER, "sha256=" + sign(payload, secret)); + } + // The delivery future is intentionally not joined: retries are scheduled from the + // completion callback and failures are only logged. + var delivery = + client + .sendAsync(requestBuilder.build(), HttpResponse.BodyHandlers.discarding()) + .whenComplete( + (response, error) -> { + Integer status = response != null ? response.statusCode() : null; + boolean succeeded = status != null && status >= 200 && status < 300; + if (succeeded) { + return; + } + if (attempt < maxAttempts) { + long delayMillis = retryBackoff.toMillis() << (attempt - 1); + LOGGER.debug( + "Webhook delivery to {} failed (attempt {}/{}, status: {}); retrying in {} ms", + endpoint, + attempt, + maxAttempts, + status, + delayMillis, + error); + var retry = + retryExecutor.schedule( + () -> send(payload, eventType, attempt + 1), + delayMillis, + TimeUnit.MILLISECONDS); + } else { + LOGGER.error( + "Webhook delivery to {} failed after {} attempt(s), dropping event" + + " (status: {})", + endpoint, + attempt, + status, + error); + } + }); + } + + private String toJson(PolarisEvent event) { + HashMap properties = new HashMap<>(); + properties.put("event_type", event.type().name()); + properties.put("timestamp", clock.millis()); + event + .attributes() + .get(EventAttributes.TABLE_IDENTIFIER) + .map(TableIdentifier::toString) + .ifPresent(id -> properties.put("table_identifier", id)); + properties.put("realm_id", event.metadata().realmId()); + event + .metadata() + .user() + .ifPresent( + p -> { + properties.put("principal", p.getName()); + properties.put("activated_roles", p.getRoles()); + }); + event.metadata().requestId().ifPresent(id -> properties.put("request_id", id)); + try { + return objectMapper.writeValueAsString(properties); + } catch (JsonProcessingException e) { + LOGGER.error("Error processing event into JSON string: ", e); + LOGGER.debug("Failed to convert the following object into JSON string: {}", properties); + return null; + } + } + + static String sign(String payload, String secret) { + try { + Mac mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256")); + byte[] digest = mac.doFinal(payload.getBytes(StandardCharsets.UTF_8)); + StringBuilder hex = new StringBuilder(digest.length * 2); + for (byte b : digest) { + hex.append(Character.forDigit((b >> 4) & 0xF, 16)); + hex.append(Character.forDigit(b & 0xF, 16)); + } + return hex.toString(); + } catch (NoSuchAlgorithmException | InvalidKeyException e) { + throw new IllegalStateException("Failed to sign webhook payload", e); + } + } +} diff --git a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java new file mode 100644 index 0000000000..b8b707129b --- /dev/null +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.extensions.events.webhook; + +import io.quarkus.runtime.annotations.StaticInitSafe; +import io.smallrye.config.ConfigMapping; +import io.smallrye.config.WithDefault; +import io.smallrye.config.WithName; +import jakarta.enterprise.context.ApplicationScoped; +import java.net.URI; +import java.time.Duration; +import java.util.Map; +import java.util.Optional; + +/** + * Configuration interface for the webhook event listener integration. + * + *

Deliveries are performed asynchronously and do not block the event executor; in-flight + * requests to a slow or unreachable endpoint are bounded by {@link #timeout()}. Delivery is + * best-effort: retries are kept in memory only, so pending retries are lost on shutdown, and an + * event that exhausts all attempts is dropped (and logged). Deployments that need stronger delivery + * guarantees should also enable the {@code persistence-in-memory-buffer} event listener. HTTP + * redirects are not followed; the endpoint must respond directly. + */ +@StaticInitSafe +@ConfigMapping(prefix = "polaris.event-listener.webhook") +@ApplicationScoped +public interface WebhookEventListenerConfiguration { + + /** + * Returns the HTTP(S) endpoint that webhook events are delivered to. + * + *

Each event is sent as an HTTP POST request with a JSON body. The endpoint must return a 2xx + * status code for the delivery to be considered successful; any other status code (or a + * connection error) triggers a retry. HTTPS is strongly recommended because the payload may + * contain sensitive audit data. + * + *

This property is required when the {@code webhook} event listener is enabled. + * + *

Configuration property: {@code polaris.event-listener.webhook.endpoint} + * + * @return an optional URI of the webhook endpoint + */ + @WithName("endpoint") + Optional endpoint(); + + /** + * Returns the shared secret used to sign webhook payloads. + * + *

When set, each request carries an {@code X-Polaris-Signature-256} header containing the hex + * encoded HMAC-SHA256 of the request body, prefixed with {@code sha256=}. Receivers should + * recompute the HMAC over the raw request body and compare it against this header to verify + * authenticity and integrity. If not set, no signature header is sent. + * + *

Configuration property: {@code polaris.event-listener.webhook.secret} + * + * @return an optional String containing the HMAC signing secret + */ + @WithName("secret") + Optional secret(); + + /** + * Returns additional HTTP headers to send with every webhook request. + * + *

This is typically used to authenticate against the receiver, for example + * `polaris.event-listener.webhook.headers."Authorization"=Bearer ` or + * `polaris.event-listener.webhook.headers."Authorization"=Splunk `. Header values may + * contain secrets; protect them like any other Polaris credential. + * + *

Configuration property: {@code polaris.event-listener.webhook.headers.""} + * + * @return a map of header names to header values + */ + @WithName("headers") + Map headers(); + + /** + * Returns the timeout for a single webhook delivery attempt. + * + *

If a delivery attempt does not complete within this duration, it is treated as failed and + * the retry schedule applies. + * + *

Configuration property: {@code polaris.event-listener.webhook.timeout} + * + * @return the delivery attempt timeout + */ + @WithName("timeout") + @WithDefault("10s") + Duration timeout(); + + /** + * Returns the maximum number of delivery attempts per event, including the initial attempt. + * + *

When all attempts fail, the event is dropped and an error is logged. + * + *

Configuration property: {@code polaris.event-listener.webhook.max-attempts} + * + * @return the maximum number of delivery attempts + */ + @WithName("max-attempts") + @WithDefault("3") + int maxAttempts(); + + /** + * Returns the initial backoff between delivery attempts. + * + *

The backoff doubles after each failed attempt (exponential backoff). + * + *

Configuration property: {@code polaris.event-listener.webhook.retry-backoff} + * + * @return the initial retry backoff + */ + @WithName("retry-backoff") + @WithDefault("1s") + Duration retryBackoff(); +} diff --git a/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java new file mode 100644 index 0000000000..dd844fbf45 --- /dev/null +++ b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java @@ -0,0 +1,245 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.extensions.events.webhook; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.when; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpServer; +import io.quarkus.runtime.configuration.MemorySize; +import java.math.BigInteger; +import java.net.InetSocketAddress; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Clock; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.service.config.PolarisIcebergObjectMapperCustomizer; +import org.apache.polaris.service.events.EventAttributeMap; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.PolarisEventType; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +class WebhookEventListenerTest { + + private static final String REALM = "test-realm"; + private static final String TEST_USER = "test-user"; + private static final String SECRET = "webhook-signing-secret"; + private static final PolarisPrincipal PRINCIPAL = + PolarisPrincipal.of(TEST_USER, Map.of(), Set.of("role1", "role2")); + private static final Clock CLOCK = Clock.systemUTC(); + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + static { + new PolarisIcebergObjectMapperCustomizer(new MemorySize(BigInteger.valueOf(1024 * 1024))) + .customize(OBJECT_MAPPER); + } + + @Mock private WebhookEventListenerConfiguration config; + + private AutoCloseable mockitoContext; + private HttpServer server; + private List recordedRequests; + private AtomicBoolean failNext; + private AtomicBoolean failAlways; + + private record RecordedRequest( + String body, String signature, String eventType, String authorization) {} + + @BeforeEach + void setUp() throws Exception { + mockitoContext = MockitoAnnotations.openMocks(this); + recordedRequests = new CopyOnWriteArrayList<>(); + failNext = new AtomicBoolean(false); + failAlways = new AtomicBoolean(false); + server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + server.createContext( + "/hook", + exchange -> { + String body = + new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8); + recordedRequests.add( + new RecordedRequest( + body, + exchange.getRequestHeaders().getFirst(WebhookEventListener.SIGNATURE_HEADER), + exchange.getRequestHeaders().getFirst(WebhookEventListener.EVENT_TYPE_HEADER), + exchange.getRequestHeaders().getFirst("Authorization"))); + int status = failAlways.get() || failNext.getAndSet(false) ? 500 : 200; + exchange.sendResponseHeaders(status, -1); + exchange.close(); + }); + server.start(); + + when(config.endpoint()).thenReturn(Optional.of(endpoint())); + when(config.secret()).thenReturn(Optional.of(SECRET)); + when(config.headers()).thenReturn(Map.of()); + when(config.timeout()).thenReturn(Duration.ofSeconds(5)); + when(config.maxAttempts()).thenReturn(3); + when(config.retryBackoff()).thenReturn(Duration.ofMillis(50)); + } + + @AfterEach + void tearDown() throws Exception { + if (server != null) { + server.stop(0); + } + if (mockitoContext != null) { + mockitoContext.close(); + } + } + + private URI endpoint() { + return URI.create("http://localhost:" + server.getAddress().getPort() + "/hook"); + } + + private WebhookEventListener createListener() { + WebhookEventListener listener = new WebhookEventListener(config, OBJECT_MAPPER, CLOCK); + listener.start(); + return listener; + } + + private static PolarisEvent testEvent() { + return new PolarisEvent( + PolarisEventType.AFTER_REFRESH_TABLE, + PolarisEventMetadata.builder().realmId(REALM).user(PRINCIPAL).build(), + new EventAttributeMap() + .put(EventAttributes.CATALOG_NAME, "test_catalog") + .put(EventAttributes.TABLE_IDENTIFIER, TableIdentifier.of("test_ns", "test_table"))); + } + + @Test + void deliversSignedJsonPayload() { + WebhookEventListener listener = createListener(); + try { + listener.onEvent(testEvent()); + + Awaitility.await("webhook delivered") + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); + + RecordedRequest request = recordedRequests.getFirst(); + assertThat(request.body()) + .contains(PolarisEventType.AFTER_REFRESH_TABLE.name()) + .contains(REALM) + .contains(TEST_USER) + .contains("test_ns.test_table"); + assertThat(request.eventType()).isEqualTo(PolarisEventType.AFTER_REFRESH_TABLE.name()); + assertThat(request.signature()) + .isEqualTo("sha256=" + WebhookEventListener.sign(request.body(), SECRET)); + } finally { + listener.shutdown(); + } + } + + @Test + void omitsSignatureHeaderWhenNoSecretConfigured() { + when(config.secret()).thenReturn(Optional.empty()); + WebhookEventListener listener = createListener(); + try { + listener.onEvent(testEvent()); + + Awaitility.await("webhook delivered") + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); + + assertThat(recordedRequests.getFirst().signature()).isNull(); + assertThat(recordedRequests.getFirst().body()).contains(REALM); + } finally { + listener.shutdown(); + } + } + + @Test + void sendsConfiguredCustomHeaders() { + when(config.headers()).thenReturn(Map.of("Authorization", "Splunk test-hec-token")); + WebhookEventListener listener = createListener(); + try { + listener.onEvent(testEvent()); + + Awaitility.await("webhook delivered") + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); + + assertThat(recordedRequests.getFirst().authorization()).isEqualTo("Splunk test-hec-token"); + } finally { + listener.shutdown(); + } + } + + @Test + void failsFastWhenEndpointNotConfigured() { + when(config.endpoint()).thenReturn(Optional.empty()); + WebhookEventListener listener = new WebhookEventListener(config, OBJECT_MAPPER, CLOCK); + assertThatThrownBy(listener::start) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("polaris.event-listener.webhook.endpoint"); + } + + @Test + void retriesFailedDeliveryUntilSuccess() { + failNext.set(true); + WebhookEventListener listener = createListener(); + try { + listener.onEvent(testEvent()); + + Awaitility.await("delivery retried and succeeded") + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(recordedRequests).hasSize(2)); + } finally { + listener.shutdown(); + } + } + + @Test + void givesUpAfterMaxAttempts() { + when(config.maxAttempts()).thenReturn(2); + failAlways.set(true); + WebhookEventListener listener = createListener(); + try { + listener.onEvent(testEvent()); + + Awaitility.await("both attempts performed") + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(recordedRequests).hasSize(2)); + + // No further attempts beyond maxAttempts. + assertThat(recordedRequests) + .allSatisfy(r -> assertThat(r.body()).isEqualTo(recordedRequests.getLast().body())); + } finally { + listener.shutdown(); + } + } +} diff --git a/gradle/projects.main.properties b/gradle/projects.main.properties index 12f6119c87..8d2fec17e3 100644 --- a/gradle/projects.main.properties +++ b/gradle/projects.main.properties @@ -52,6 +52,7 @@ polaris-extensions-federation-hive=extensions/federation/hive polaris-extensions-federation-bigquery=extensions/federation/bigquery polaris-extensions-auth-opa=extensions/auth/opa polaris-extensions-auth-ranger=extensions/auth/ranger +polaris-extensions-events-webhook=extensions/events/webhook polaris-extensions-semantic-models=extensions/semantic-models polaris-config-docs-annotations=tools/config-docs/annotations diff --git a/runtime/defaults/src/main/resources/application.properties b/runtime/defaults/src/main/resources/application.properties index dc63be8705..1e7eef5f9d 100644 --- a/runtime/defaults/src/main/resources/application.properties +++ b/runtime/defaults/src/main/resources/application.properties @@ -173,6 +173,15 @@ polaris.file-io.type=default # polaris.event-listener.aws-cloudwatch.region=us-east-1 # polaris.event-listener.aws-cloudwatch.synchronous-mode=false +# Webhook event listener settings +# polaris.event-listener.types=webhook +# polaris.event-listener.webhook.endpoint=https://example.com/polaris-events +# polaris.event-listener.webhook.secret=change-me +# polaris.event-listener.webhook.headers."Authorization"=Bearer change-me +# polaris.event-listener.webhook.timeout=10s +# polaris.event-listener.webhook.max-attempts=3 +# polaris.event-listener.webhook.retry-backoff=1s + # OpenTelemetry event listener settings # polaris.event-listener.types=opentelemetry # quarkus.otel.sdk.disabled=false diff --git a/runtime/server/build.gradle.kts b/runtime/server/build.gradle.kts index 943394567b..eed4b65664 100644 --- a/runtime/server/build.gradle.kts +++ b/runtime/server/build.gradle.kts @@ -41,6 +41,7 @@ dependencies { runtimeOnly(project(":polaris-extensions-federation-hadoop")) runtimeOnly(project(":polaris-extensions-auth-opa")) runtimeOnly(project(":polaris-extensions-auth-ranger")) + runtimeOnly(project(":polaris-extensions-events-webhook")) runtimeOnly(project(":polaris-extensions-semantic-models")) val nonRestCatalogs = providers.gradleProperty("NonRESTCatalogs").orNull diff --git a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md new file mode 100644 index 0000000000..cf9fa36e40 --- /dev/null +++ b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md @@ -0,0 +1,37 @@ +--- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +title: smallrye-polaris_event_listener_webhook +build: + list: never + render: never +--- + +Configuration interface for the webhook event listener integration. + +Deliveries are performed asynchronously and do not block the event executor; in-flight requests to a slow or unreachable endpoint are bounded by (`#timeout()`). Delivery is best-effort: retries are kept in memory only, so pending retries are lost on shutdown, and an event that exhausts all attempts is dropped (and logged). Deployments that need stronger delivery guarantees should also enable the `persistence-in-memory-buffer` event listener. HTTP redirects are not followed; the endpoint must respond directly. + +| Property | Default Value | Type | Description | +|----------|---------------|------|-------------| +| `polaris.event-listener.webhook.endpoint` | | `uri` | Returns the HTTP(S) endpoint that webhook events are delivered to.

Each event is sent as an HTTP POST request with a JSON body. The endpoint must return a 2xx status code for the delivery to be considered successful; any other status code (or a connection error) triggers a retry. HTTPS is strongly recommended because the payload may contain sensitive audit data.

This property is required when the `webhook` event listener is enabled.

Configuration property: `polaris.event-listener.webhook.endpoint` | +| `polaris.event-listener.webhook.secret` | | `string` | Returns the shared secret used to sign webhook payloads.

When set, each request carries an `X-Polaris-Signature-256` header containing the hex encoded HMAC-SHA256 of the request body, prefixed with `sha256=`. Receivers should recompute the HMAC over the raw request body and compare it against this header to verify authenticity and integrity. If not set, no signature header is sent.

Configuration property: `polaris.event-listener.webhook.secret` | +| `polaris.event-listener.webhook.headers.`_``_ | | `string` | Returns additional HTTP headers to send with every webhook request.

This is typically used to authenticate against the receiver, for example `polaris.event-listener.webhook.headers."Authorization"=Bearer ` or `polaris.event-listener.webhook.headers."Authorization"=Splunk `. Header values may contain secrets; protect them like any other Polaris credential.

Configuration property: `polaris.event-listener.webhook.headers.""` | +| `polaris.event-listener.webhook.timeout` | `10s` | `duration` | Returns the timeout for a single webhook delivery attempt.

If a delivery attempt does not complete within this duration, it is treated as failed and the retry schedule applies.

Configuration property: `polaris.event-listener.webhook.timeout` | +| `polaris.event-listener.webhook.max-attempts` | `3` | `int` | Returns the maximum number of delivery attempts per event, including the initial attempt.

When all attempts fail, the event is dropped and an error is logged.

Configuration property: `polaris.event-listener.webhook.max-attempts` | +| `polaris.event-listener.webhook.retry-backoff` | `1s` | `duration` | Returns the initial backoff between delivery attempts.

The backoff doubles after each failed attempt (exponential backoff).

Configuration property: `polaris.event-listener.webhook.retry-backoff` | diff --git a/site/content/in-dev/unreleased/configuration/configuration-reference.md b/site/content/in-dev/unreleased/configuration/configuration-reference.md index b010b469cc..ef79098a51 100644 --- a/site/content/in-dev/unreleased/configuration/configuration-reference.md +++ b/site/content/in-dev/unreleased/configuration/configuration-reference.md @@ -138,6 +138,36 @@ All properties listed here are **runtime** properties and can be changed without {{% include-config-section "smallrye-polaris_event_listener_aws_cloudwatch" %}} +### `polaris.event-listener.webhook` + +{{% include-config-section "smallrye-polaris_event_listener_webhook" %}} + +The webhook listener delivers each (sanitized) event as an HTTP POST with a JSON body of the +following shape: + +```json +{ + "event_type": "AFTER_REFRESH_TABLE", + "timestamp": 1760000000000, + "realm_id": "my-realm", + "principal": "alice", + "activated_roles": ["role1", "role2"], + "request_id": "d3b0...", + "table_identifier": "ns.tbl" +} +``` + +`principal`, `activated_roles`, `request_id`, and `table_identifier` are omitted when not +applicable to the event. Each request also carries an `X-Polaris-Event` header with the event +type. When `polaris.event-listener.webhook.secret` is set, an `X-Polaris-Signature-256` header +contains `sha256=` followed by the hex-encoded HMAC-SHA256 of the raw request body; receivers +should recompute the HMAC with the shared secret and compare. + +Delivery is best-effort: failed deliveries are retried with exponential backoff and dropped after +`max-attempts`. Retries are in-memory only, so pending retries are lost on restart. Deployments +that need stronger delivery guarantees should also enable the `persistence-in-memory-buffer` +listener. Redirects are not followed; the endpoint must respond directly with a 2xx status code. + ### `opentelemetry` event listener Set `polaris.event-listener.types=opentelemetry` to emit Polaris events as OpenTelemetry log diff --git a/tools/config-docs/site/build.gradle.kts b/tools/config-docs/site/build.gradle.kts index ff72a40008..1b21c2712b 100644 --- a/tools/config-docs/site/build.gradle.kts +++ b/tools/config-docs/site/build.gradle.kts @@ -29,6 +29,7 @@ val genProjectPaths = listOf( ":polaris-async-api", ":polaris-core", ":polaris-extensions-auth-opa", + ":polaris-extensions-events-webhook", ":polaris-nodes-api", ":polaris-persistence-nosql-api", ":polaris-persistence-nosql-impl", From 88eb592595fda569cb2adb2f7dfc3a54b3fc516c Mon Sep 17 00:00:00 2001 From: vignesh_A Date: Mon, 20 Jul 2026 18:40:26 +0530 Subject: [PATCH 2/5] feat(events): harden webhook listener (envelope, bounds, retry, metrics) Address review feedback for the webhook event listener: versioned JSON envelope with event id and original timestamp; bounded pending and concurrency; transient-only retries with jitter and Retry-After; HTTPS by default and reserved header protection; Micrometer metrics; honest best-effort documentation. --- CHANGELOG.md | 2 +- extensions/events/webhook/build.gradle.kts | 2 + .../events/webhook/WebhookEventListener.java | 414 +++++++++++++++--- .../WebhookEventListenerConfiguration.java | 108 +++-- .../webhook/WebhookEventListenerTest.java | 207 ++++++++- .../src/main/resources/application.properties | 4 + ...smallrye-polaris_event_listener_webhook.md | 27 +- .../configuration/configuration-reference.md | 29 +- 8 files changed, 656 insertions(+), 137 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a1dd037d56..78689e14c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,7 +48,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti - Added optional `sessionPolicy` field to `SigV4AuthenticationParameters` for catalog federation. When set, the IAM session policy JSON is attached to the STS AssumeRole request, allowing administrators to restrict vended credentials to only the required AWS services and actions (Principle of Least Privilege). - Python CLI: added `--scheme` to specify URL scheme - Added opt-in idempotency for `createTable` and `updateTable` in the Iceberg REST catalog. When enabled via `polaris.idempotency.enabled=true` (default `false`), a client-supplied `Idempotency-Key` header is embedded into the table entity and committed in the same transaction as the operation; a retry carrying the same key within the TTL window (`polaris.idempotency.ttl`, default `PT5M`) replays the original success instead of failing — with `AlreadyExists` for `createTable`, or with `CommitFailedException` for `updateTable` when the request's requirements no longer match the already-advanced table. -- Added a `webhook` event listener (`polaris.event-listener.types=webhook`) that delivers Polaris events as JSON HTTP POST requests to a configurable endpoint, for example a SIEM or audit collector. Deliveries are retried with exponential backoff (`polaris.event-listener.webhook.max-attempts`, `polaris.event-listener.webhook.retry-backoff`) on a best-effort basis, payloads can be signed with HMAC-SHA256 via `polaris.event-listener.webhook.secret` (sent as the `X-Polaris-Signature-256` header) so receivers can verify authenticity, and `polaris.event-listener.webhook.headers` allows custom headers such as `Authorization` for receivers that require them. +- Added a `webhook` event listener (`polaris.event-listener.types=webhook`) that delivers sanitized Polaris events as JSON HTTP POSTs to a configurable endpoint (SIEM/audit collectors). Payloads are a versioned envelope (`schema_version`, `event_id`, original `timestamp`, optional `delivery_timestamp`) with optional HMAC-SHA256 signing (`X-Polaris-Signature-256`). Delivery is best-effort and bounded (`max-concurrent`, `max-pending`); transient failures retry with jitter (and `Retry-After` when present); HTTPS is required by default (`require-https`); reserved signature/event headers cannot be overridden by custom headers. Metrics cover success, failure, retries, drops, pending, in-flight, and latency. ### Changes - The admin tool's `bootstrap` command is now idempotent: bootstrapping a realm that is already diff --git a/extensions/events/webhook/build.gradle.kts b/extensions/events/webhook/build.gradle.kts index b1edb5b15d..0ed45dbc04 100644 --- a/extensions/events/webhook/build.gradle.kts +++ b/extensions/events/webhook/build.gradle.kts @@ -35,6 +35,8 @@ dependencies { implementation(platform(libs.quarkus.bom)) implementation("io.quarkus:quarkus-core") implementation("io.quarkus:quarkus-jackson") + implementation("io.micrometer:micrometer-core") + implementation(libs.guava) compileOnly(libs.jakarta.enterprise.cdi.api) compileOnly(libs.jakarta.inject.api) diff --git a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java index 1b9899db5a..be882b0626 100644 --- a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java @@ -20,6 +20,11 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.annotations.VisibleForTesting; +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.Gauge; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Timer; import io.smallrye.common.annotation.Identifier; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; @@ -27,18 +32,26 @@ import jakarta.inject.Inject; import java.net.URI; import java.net.http.HttpClient; +import java.net.http.HttpHeaders; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; import java.security.InvalidKeyException; import java.security.NoSuchAlgorithmException; +import java.security.SecureRandom; import java.time.Clock; import java.time.Duration; import java.util.HashMap; +import java.util.Locale; import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import javax.crypto.Mac; import javax.crypto.spec.SecretKeySpec; import org.apache.iceberg.catalog.TableIdentifier; @@ -49,9 +62,11 @@ import org.slf4j.LoggerFactory; /** - * Delivers Polaris events to an HTTP(S) endpoint as JSON POST requests, optionally signed with - * HMAC-SHA256. Failed deliveries (non-2xx responses or connection errors) are retried with - * exponential backoff; events that exhaust all attempts are dropped and logged. + * Delivers Polaris events to an HTTP(S) endpoint as JSON POSTs, optionally signed with HMAC-SHA256. + * + *

Delivery is best-effort and bounded: concurrency and pending work are capped; retries use + * exponential backoff with full jitter for transient failures only. Ordering is not guaranteed. + * In-flight and pending work are not durable across restarts. */ @ApplicationScoped @Identifier("webhook") @@ -60,6 +75,15 @@ public class WebhookEventListener implements PolarisEventListener { static final String SIGNATURE_HEADER = "X-Polaris-Signature-256"; static final String EVENT_TYPE_HEADER = "X-Polaris-Event"; + static final int SCHEMA_VERSION = 1; + + private static final Set RESERVED_HEADERS = + Set.of( + "content-type", + "content-length", + "host", + SIGNATURE_HEADER.toLowerCase(Locale.ROOT), + EVENT_TYPE_HEADER.toLowerCase(Locale.ROOT)); private final URI endpoint; private final String secret; @@ -67,23 +91,48 @@ public class WebhookEventListener implements PolarisEventListener { private final Duration timeout; private final int maxAttempts; private final Duration retryBackoff; + private final Duration maxRetryBackoff; + private final int maxConcurrent; + private final int maxPending; + private final boolean requireHttps; private final ObjectMapper objectMapper; private final Clock clock; + private final MeterRegistry meterRegistry; + private final SecureRandom random = new SecureRandom(); + + private final AtomicInteger pending = new AtomicInteger(); + private final AtomicInteger inFlight = new AtomicInteger(); private HttpClient client; + private ExecutorService deliveryExecutor; private ScheduledExecutorService retryExecutor; + private Semaphore concurrency; + + private Counter successCounter; + private Counter failureCounter; + private Counter retryCounter; + private Counter dropCounter; + private Timer deliveryTimer; @Inject public WebhookEventListener( - WebhookEventListenerConfiguration config, ObjectMapper objectMapper, Clock clock) { + WebhookEventListenerConfiguration config, + ObjectMapper objectMapper, + Clock clock, + MeterRegistry meterRegistry) { this.endpoint = config.endpoint().orElse(null); this.secret = config.secret().orElse(null); - this.headers = config.headers(); + this.headers = Map.copyOf(config.headers()); this.timeout = config.timeout(); this.maxAttempts = config.maxAttempts(); this.retryBackoff = config.retryBackoff(); + this.maxRetryBackoff = config.maxRetryBackoff(); + this.maxConcurrent = Math.max(1, config.maxConcurrent()); + this.maxPending = Math.max(1, config.maxPending()); + this.requireHttps = config.requireHttps(); this.objectMapper = objectMapper; this.clock = clock; + this.meterRegistry = meterRegistry; } @PostConstruct @@ -93,7 +142,17 @@ void start() { "The webhook event listener is enabled but polaris.event-listener.webhook.endpoint is" + " not set"); } + validateEndpoint(endpoint, requireHttps); this.client = createHttpClient(); + this.concurrency = new Semaphore(maxConcurrent); + this.deliveryExecutor = + Executors.newFixedThreadPool( + maxConcurrent, + runnable -> { + Thread thread = new Thread(runnable, "polaris-webhook-event-listener-delivery"); + thread.setDaemon(true); + return thread; + }); this.retryExecutor = Executors.newSingleThreadScheduledExecutor( runnable -> { @@ -101,10 +160,68 @@ void start() { thread.setDaemon(true); return thread; }); + registerMetrics(); + } + + private void registerMetrics() { + successCounter = + Counter.builder("polaris.event.webhook.deliveries") + .tag("result", "success") + .description("Successful webhook deliveries") + .register(meterRegistry); + failureCounter = + Counter.builder("polaris.event.webhook.deliveries") + .tag("result", "failure") + .description("Failed webhook deliveries after retries exhausted or permanent errors") + .register(meterRegistry); + retryCounter = + Counter.builder("polaris.event.webhook.retries") + .description("Scheduled webhook delivery retries") + .register(meterRegistry); + dropCounter = + Counter.builder("polaris.event.webhook.drops") + .description("Events dropped (queue full or non-retryable failure)") + .register(meterRegistry); + deliveryTimer = + Timer.builder("polaris.event.webhook.delivery") + .description("Webhook HTTP attempt latency") + .register(meterRegistry); + Gauge.builder("polaris.event.webhook.pending", pending, AtomicInteger::get) + .description("Events accepted but not yet finished (in flight or awaiting retry)") + .register(meterRegistry); + Gauge.builder("polaris.event.webhook.in_flight", inFlight, AtomicInteger::get) + .description("Concurrent HTTP attempts in progress") + .register(meterRegistry); + } + + @VisibleForTesting + static void validateEndpoint(URI endpoint, boolean requireHttps) { + String scheme = endpoint.getScheme(); + if (scheme == null) { + throw new IllegalStateException( + "polaris.event-listener.webhook.endpoint must include a scheme (https://...)"); + } + String lower = scheme.toLowerCase(Locale.ROOT); + if (requireHttps && !"https".equals(lower)) { + throw new IllegalStateException( + "polaris.event-listener.webhook.endpoint must use https when require-https is true" + + " (got scheme '" + + scheme + + "')"); + } + if (!"https".equals(lower) && !"http".equals(lower)) { + throw new IllegalStateException( + "polaris.event-listener.webhook.endpoint must use http or https (got scheme '" + + scheme + + "')"); + } } protected HttpClient createHttpClient() { - return HttpClient.newBuilder().connectTimeout(timeout).build(); + return HttpClient.newBuilder() + .connectTimeout(timeout) + .followRedirects(HttpClient.Redirect.NEVER) + .build(); } @PreDestroy @@ -113,71 +230,254 @@ void shutdown() { retryExecutor.shutdownNow(); retryExecutor = null; } + if (deliveryExecutor != null) { + deliveryExecutor.shutdownNow(); + deliveryExecutor = null; + } } @Override public void onEvent(PolarisEvent event) { + if (!tryAccept()) { + dropCounter.increment(); + LOGGER.warn( + "Webhook pending capacity full ({}); dropping event type={}", + maxPending, + event.type().name()); + return; + } String payload = toJson(event); if (payload == null) { + releaseAccepted(); + dropCounter.increment(); + return; + } + scheduleDelivery(payload, event.type().name(), 1, /*delayMs*/ 0); + } + + /** Try to reserve a pending slot for a new event. */ + private boolean tryAccept() { + for (; ; ) { + int current = pending.get(); + if (current >= maxPending) { + return false; + } + if (pending.compareAndSet(current, current + 1)) { + return true; + } + } + } + + private void releaseAccepted() { + pending.decrementAndGet(); + } + + private void scheduleDelivery(String payload, String eventType, int attempt, long delayMs) { + ExecutorService delivery = deliveryExecutor; + ScheduledExecutorService retry = retryExecutor; + if (delivery == null || delivery.isShutdown()) { + releaseAccepted(); + dropCounter.increment(); + return; + } + Runnable task = () -> deliverWithConcurrency(payload, eventType, attempt); + if (delayMs <= 0) { + try { + delivery.execute(task); + } catch (RuntimeException e) { + releaseAccepted(); + dropCounter.increment(); + } + return; + } + if (retry == null || retry.isShutdown()) { + releaseAccepted(); + dropCounter.increment(); + return; + } + // Retries are best-effort; the Future is not joined (delivery completion is handled in-task). + try { + var unused = + retry.schedule( + () -> { + ExecutorService d = deliveryExecutor; + if (d == null || d.isShutdown()) { + releaseAccepted(); + dropCounter.increment(); + return; + } + try { + d.execute(task); + } catch (RuntimeException e) { + releaseAccepted(); + dropCounter.increment(); + } + }, + delayMs, + TimeUnit.MILLISECONDS); + } catch (RuntimeException e) { + releaseAccepted(); + dropCounter.increment(); + } + } + + private void deliverWithConcurrency(String payload, String eventType, int attempt) { + try { + concurrency.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + releaseAccepted(); + dropCounter.increment(); + LOGGER.warn("Interrupted while waiting for webhook concurrency; dropping event"); + return; + } + inFlight.incrementAndGet(); + long startNanos = System.nanoTime(); + try { + HttpRequest request = buildRequest(payload, eventType); + HttpResponse response = client.send(request, HttpResponse.BodyHandlers.discarding()); + deliveryTimer.record(System.nanoTime() - startNanos, TimeUnit.NANOSECONDS); + int status = response.statusCode(); + if (status >= 200 && status < 300) { + successCounter.increment(); + releaseAccepted(); + return; + } + handleFailure(payload, eventType, attempt, status, response.headers(), null); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + deliveryTimer.record(System.nanoTime() - startNanos, TimeUnit.NANOSECONDS); + handleFailure(payload, eventType, attempt, null, null, e); + } catch (Exception e) { + deliveryTimer.record(System.nanoTime() - startNanos, TimeUnit.NANOSECONDS); + handleFailure(payload, eventType, attempt, null, null, e); + } finally { + inFlight.decrementAndGet(); + concurrency.release(); + } + } + + private void handleFailure( + String payload, + String eventType, + int attempt, + Integer status, + HttpHeaders headers, + Throwable error) { + boolean transientFailure = isTransient(status, error); + if (transientFailure && attempt < maxAttempts) { + long delayMs = computeRetryDelayMillis(attempt, headers); + retryCounter.increment(); + LOGGER.debug( + "Webhook delivery to {} failed (attempt {}/{}, status: {}); retrying in {} ms", + endpoint, + attempt, + maxAttempts, + status, + delayMs, + error); + scheduleDelivery(payload, eventType, attempt + 1, delayMs); return; } - send(payload, event.type().name(), 1); + failureCounter.increment(); + releaseAccepted(); + LOGGER.error( + "Webhook delivery to {} failed after {} attempt(s), dropping event (status: {}, transient:" + + " {})", + endpoint, + attempt, + status, + transientFailure, + error); + } + + @VisibleForTesting + static boolean isTransient(Integer status, Throwable error) { + if (error != null) { + return true; + } + if (status == null) { + return true; + } + if (status == 408 || status == 425 || status == 429) { + return true; + } + return status >= 500 && status <= 599; + } + + @VisibleForTesting + long computeRetryDelayMillis(int attempt, HttpHeaders headers) { + Optional retryAfter = parseRetryAfterMillis(headers); + if (retryAfter.isPresent()) { + return Math.min(retryAfter.get(), maxRetryBackoff.toMillis()); + } + long exp = retryBackoff.toMillis() << Math.min(attempt - 1, 16); + long capped = Math.min(exp, maxRetryBackoff.toMillis()); + // Full jitter: random in [0, capped] + if (capped <= 0) { + return 0; + } + return (long) (random.nextDouble() * capped); + } + + @VisibleForTesting + static Optional parseRetryAfterMillis(HttpHeaders headers) { + if (headers == null) { + return Optional.empty(); + } + return headers + .firstValue("Retry-After") + .map(String::trim) + .flatMap( + value -> { + try { + // Delta-seconds form only (HTTP-date not supported). + long seconds = Long.parseLong(value); + if (seconds < 0) { + return Optional.empty(); + } + return Optional.of(TimeUnit.SECONDS.toMillis(seconds)); + } catch (NumberFormatException e) { + return Optional.empty(); + } + }); } - private void send(String payload, String eventType, int attempt) { + private HttpRequest buildRequest(String payload, String eventType) { HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(endpoint) .timeout(timeout) .header("Content-Type", "application/json") - .header(EVENT_TYPE_HEADER, eventType); - headers.forEach(requestBuilder::header); - requestBuilder.POST(HttpRequest.BodyPublishers.ofString(payload, StandardCharsets.UTF_8)); + .header(EVENT_TYPE_HEADER, eventType) + .POST(HttpRequest.BodyPublishers.ofString(payload, StandardCharsets.UTF_8)); + headers.forEach( + (name, value) -> { + if (name != null && !isReservedHeader(name)) { + requestBuilder.header(name, value); + } else if (name != null) { + LOGGER.debug("Ignoring reserved custom webhook header '{}'", name); + } + }); if (secret != null) { requestBuilder.header(SIGNATURE_HEADER, "sha256=" + sign(payload, secret)); } - // The delivery future is intentionally not joined: retries are scheduled from the - // completion callback and failures are only logged. - var delivery = - client - .sendAsync(requestBuilder.build(), HttpResponse.BodyHandlers.discarding()) - .whenComplete( - (response, error) -> { - Integer status = response != null ? response.statusCode() : null; - boolean succeeded = status != null && status >= 200 && status < 300; - if (succeeded) { - return; - } - if (attempt < maxAttempts) { - long delayMillis = retryBackoff.toMillis() << (attempt - 1); - LOGGER.debug( - "Webhook delivery to {} failed (attempt {}/{}, status: {}); retrying in {} ms", - endpoint, - attempt, - maxAttempts, - status, - delayMillis, - error); - var retry = - retryExecutor.schedule( - () -> send(payload, eventType, attempt + 1), - delayMillis, - TimeUnit.MILLISECONDS); - } else { - LOGGER.error( - "Webhook delivery to {} failed after {} attempt(s), dropping event" - + " (status: {})", - endpoint, - attempt, - status, - error); - } - }); - } - - private String toJson(PolarisEvent event) { + return requestBuilder.build(); + } + + @VisibleForTesting + static boolean isReservedHeader(String name) { + return RESERVED_HEADERS.contains(name.toLowerCase(Locale.ROOT)); + } + + @VisibleForTesting + String toJson(PolarisEvent event) { HashMap properties = new HashMap<>(); + properties.put("schema_version", SCHEMA_VERSION); + properties.put("event_id", event.metadata().eventId().toString()); properties.put("event_type", event.type().name()); - properties.put("timestamp", clock.millis()); + // Original event time from metadata (not delivery time). + properties.put("timestamp", event.metadata().timestamp().toEpochMilli()); + properties.put("delivery_timestamp", clock.millis()); event .attributes() .get(EventAttributes.TABLE_IDENTIFIER) @@ -217,4 +517,14 @@ static String sign(String payload, String secret) { throw new IllegalStateException("Failed to sign webhook payload", e); } } + + @VisibleForTesting + int pendingCount() { + return pending.get(); + } + + @VisibleForTesting + MeterRegistry meterRegistry() { + return meterRegistry; + } } diff --git a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java index b8b707129b..3b862dcb6f 100644 --- a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java @@ -29,14 +29,16 @@ import java.util.Optional; /** - * Configuration interface for the webhook event listener integration. + * Configuration for the webhook event listener. * - *

Deliveries are performed asynchronously and do not block the event executor; in-flight - * requests to a slow or unreachable endpoint are bounded by {@link #timeout()}. Delivery is - * best-effort: retries are kept in memory only, so pending retries are lost on shutdown, and an - * event that exhausts all attempts is dropped (and logged). Deployments that need stronger delivery - * guarantees should also enable the {@code persistence-in-memory-buffer} event listener. HTTP - * redirects are not followed; the endpoint must respond directly. + *

Delivery is best-effort and in-process only: pending and in-flight deliveries + * are not durable and are lost on process restart; events that exhaust retries or cannot be + * accepted into the bounded queue are dropped. Ordering is not guaranteed. This listener is not a + * durable audit spool and is not strengthened by enabling {@code persistence-in-memory-buffer} + * (that buffer is a separate listener, not a webhook delivery queue). + * + *

HTTP redirects are not followed; the endpoint must respond directly. When {@link + * #requireHttps()} is {@code true} (the default), only {@code https} endpoints are accepted. */ @StaticInitSafe @ConfigMapping(prefix = "polaris.event-listener.webhook") @@ -44,89 +46,101 @@ public interface WebhookEventListenerConfiguration { /** - * Returns the HTTP(S) endpoint that webhook events are delivered to. + * HTTP(S) endpoint that receives webhook POSTs with a JSON body. * - *

Each event is sent as an HTTP POST request with a JSON body. The endpoint must return a 2xx - * status code for the delivery to be considered successful; any other status code (or a - * connection error) triggers a retry. HTTPS is strongly recommended because the payload may - * contain sensitive audit data. + *

A 2xx response marks success. Transient failures (network errors, 408/429/5xx) are retried + * within the concurrency limits; permanent client errors (other 4xx) are not retried. * - *

This property is required when the {@code webhook} event listener is enabled. + *

Required when the {@code webhook} event listener is enabled. * *

Configuration property: {@code polaris.event-listener.webhook.endpoint} - * - * @return an optional URI of the webhook endpoint */ @WithName("endpoint") Optional endpoint(); /** - * Returns the shared secret used to sign webhook payloads. - * - *

When set, each request carries an {@code X-Polaris-Signature-256} header containing the hex - * encoded HMAC-SHA256 of the request body, prefixed with {@code sha256=}. Receivers should - * recompute the HMAC over the raw request body and compare it against this header to verify - * authenticity and integrity. If not set, no signature header is sent. + * Shared secret for HMAC-SHA256 signing. When set, each request includes {@code + * X-Polaris-Signature-256: sha256=} over the raw body. HMAC authenticates and protects + * integrity; it does not encrypt the payload — use HTTPS for confidentiality. * *

Configuration property: {@code polaris.event-listener.webhook.secret} - * - * @return an optional String containing the HMAC signing secret */ @WithName("secret") Optional secret(); /** - * Returns additional HTTP headers to send with every webhook request. - * - *

This is typically used to authenticate against the receiver, for example - * `polaris.event-listener.webhook.headers."Authorization"=Bearer ` or - * `polaris.event-listener.webhook.headers."Authorization"=Splunk `. Header values may - * contain secrets; protect them like any other Polaris credential. + * Extra headers on every request (for example {@code Authorization}). Reserved headers ({@code + * Content-Type}, {@code X-Polaris-Signature-256}, {@code X-Polaris-Event}, {@code Host}, {@code + * Content-Length}) cannot be overridden. Header values may be secrets; protect them accordingly. * *

Configuration property: {@code polaris.event-listener.webhook.headers.""} - * - * @return a map of header names to header values */ @WithName("headers") Map headers(); /** - * Returns the timeout for a single webhook delivery attempt. - * - *

If a delivery attempt does not complete within this duration, it is treated as failed and - * the retry schedule applies. + * Timeout for a single HTTP attempt. Timed-out attempts count as transient failures. * *

Configuration property: {@code polaris.event-listener.webhook.timeout} - * - * @return the delivery attempt timeout */ @WithName("timeout") @WithDefault("10s") Duration timeout(); /** - * Returns the maximum number of delivery attempts per event, including the initial attempt. - * - *

When all attempts fail, the event is dropped and an error is logged. + * Maximum delivery attempts per event including the first try. After this, the event is dropped. * *

Configuration property: {@code polaris.event-listener.webhook.max-attempts} - * - * @return the maximum number of delivery attempts */ @WithName("max-attempts") @WithDefault("3") int maxAttempts(); /** - * Returns the initial backoff between delivery attempts. - * - *

The backoff doubles after each failed attempt (exponential backoff). + * Base delay before the first retry. Subsequent delays grow exponentially with full jitter, + * capped by {@link #maxRetryBackoff()}. * *

Configuration property: {@code polaris.event-listener.webhook.retry-backoff} - * - * @return the initial retry backoff */ @WithName("retry-backoff") @WithDefault("1s") Duration retryBackoff(); + + /** + * Upper bound on retry delay after exponential growth and jitter. + * + *

Configuration property: {@code polaris.event-listener.webhook.max-retry-backoff} + */ + @WithName("max-retry-backoff") + @WithDefault("1m") + Duration maxRetryBackoff(); + + /** + * Maximum concurrent HTTP deliveries (including retries currently executing). Additional work + * waits on the concurrency limit while still counting toward {@link #maxPending()}. + * + *

Configuration property: {@code polaris.event-listener.webhook.max-concurrent} + */ + @WithName("max-concurrent") + @WithDefault("16") + int maxConcurrent(); + + /** + * Maximum events accepted for delivery or retry at once. When full, new events are dropped. + * + *

Configuration property: {@code polaris.event-listener.webhook.max-pending} + */ + @WithName("max-pending") + @WithDefault("1000") + int maxPending(); + + /** + * When {@code true}, only {@code https} endpoints are allowed (recommended for production). Set + * {@code false} only for local testing with plain HTTP. + * + *

Configuration property: {@code polaris.event-listener.webhook.require-https} + */ + @WithName("require-https") + @WithDefault("true") + boolean requireHttps(); } diff --git a/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java index dd844fbf45..5d4e1cc38b 100644 --- a/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java +++ b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java @@ -22,21 +22,27 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.when; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.sun.net.httpserver.HttpServer; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import io.quarkus.runtime.configuration.MemorySize; import java.math.BigInteger; import java.net.InetSocketAddress; import java.net.URI; +import java.net.http.HttpHeaders; import java.nio.charset.StandardCharsets; import java.time.Clock; import java.time.Duration; +import java.time.Instant; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.UUID; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.polaris.core.auth.PolarisPrincipal; import org.apache.polaris.service.config.PolarisIcebergObjectMapperCustomizer; @@ -57,6 +63,8 @@ class WebhookEventListenerTest { private static final String REALM = "test-realm"; private static final String TEST_USER = "test-user"; private static final String SECRET = "webhook-signing-secret"; + private static final UUID EVENT_ID = UUID.fromString("11111111-2222-3333-4444-555555555555"); + private static final Instant EVENT_TIME = Instant.parse("2026-01-15T12:00:00Z"); private static final PolarisPrincipal PRINCIPAL = PolarisPrincipal.of(TEST_USER, Map.of(), Set.of("role1", "role2")); private static final Clock CLOCK = Clock.systemUTC(); @@ -75,6 +83,8 @@ class WebhookEventListenerTest { private List recordedRequests; private AtomicBoolean failNext; private AtomicBoolean failAlways; + private AtomicInteger statusToReturn; + private SimpleMeterRegistry meterRegistry; private record RecordedRequest( String body, String signature, String eventType, String authorization) {} @@ -85,6 +95,8 @@ void setUp() throws Exception { recordedRequests = new CopyOnWriteArrayList<>(); failNext = new AtomicBoolean(false); failAlways = new AtomicBoolean(false); + statusToReturn = new AtomicInteger(200); + meterRegistry = new SimpleMeterRegistry(); server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); server.createContext( "/hook", @@ -97,7 +109,14 @@ void setUp() throws Exception { exchange.getRequestHeaders().getFirst(WebhookEventListener.SIGNATURE_HEADER), exchange.getRequestHeaders().getFirst(WebhookEventListener.EVENT_TYPE_HEADER), exchange.getRequestHeaders().getFirst("Authorization"))); - int status = failAlways.get() || failNext.getAndSet(false) ? 500 : 200; + int status; + if (failAlways.get()) { + status = 500; + } else if (failNext.getAndSet(false)) { + status = 500; + } else { + status = statusToReturn.get(); + } exchange.sendResponseHeaders(status, -1); exchange.close(); }); @@ -108,7 +127,11 @@ void setUp() throws Exception { when(config.headers()).thenReturn(Map.of()); when(config.timeout()).thenReturn(Duration.ofSeconds(5)); when(config.maxAttempts()).thenReturn(3); - when(config.retryBackoff()).thenReturn(Duration.ofMillis(50)); + when(config.retryBackoff()).thenReturn(Duration.ofMillis(20)); + when(config.maxRetryBackoff()).thenReturn(Duration.ofSeconds(5)); + when(config.maxConcurrent()).thenReturn(4); + when(config.maxPending()).thenReturn(100); + when(config.requireHttps()).thenReturn(false); } @AfterEach @@ -126,7 +149,8 @@ private URI endpoint() { } private WebhookEventListener createListener() { - WebhookEventListener listener = new WebhookEventListener(config, OBJECT_MAPPER, CLOCK); + WebhookEventListener listener = + new WebhookEventListener(config, OBJECT_MAPPER, CLOCK, meterRegistry); listener.start(); return listener; } @@ -134,14 +158,19 @@ private WebhookEventListener createListener() { private static PolarisEvent testEvent() { return new PolarisEvent( PolarisEventType.AFTER_REFRESH_TABLE, - PolarisEventMetadata.builder().realmId(REALM).user(PRINCIPAL).build(), + PolarisEventMetadata.builder() + .eventId(EVENT_ID) + .timestamp(EVENT_TIME) + .realmId(REALM) + .user(PRINCIPAL) + .build(), new EventAttributeMap() .put(EventAttributes.CATALOG_NAME, "test_catalog") .put(EventAttributes.TABLE_IDENTIFIER, TableIdentifier.of("test_ns", "test_table"))); } @Test - void deliversSignedJsonPayload() { + void deliversSignedJsonPayloadWithEnvelopeFields() throws Exception { WebhookEventListener listener = createListener(); try { listener.onEvent(testEvent()); @@ -151,14 +180,26 @@ void deliversSignedJsonPayload() { .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); RecordedRequest request = recordedRequests.getFirst(); - assertThat(request.body()) - .contains(PolarisEventType.AFTER_REFRESH_TABLE.name()) - .contains(REALM) - .contains(TEST_USER) - .contains("test_ns.test_table"); + JsonNode json = OBJECT_MAPPER.readTree(request.body()); + assertThat(json.get("schema_version").asInt()).isEqualTo(WebhookEventListener.SCHEMA_VERSION); + assertThat(json.get("event_id").asText()).isEqualTo(EVENT_ID.toString()); + assertThat(json.get("timestamp").asLong()).isEqualTo(EVENT_TIME.toEpochMilli()); + assertThat(json.get("delivery_timestamp").asLong()).isPositive(); + assertThat(json.get("event_type").asText()) + .isEqualTo(PolarisEventType.AFTER_REFRESH_TABLE.name()); + assertThat(json.get("realm_id").asText()).isEqualTo(REALM); + assertThat(json.get("principal").asText()).isEqualTo(TEST_USER); + assertThat(json.get("table_identifier").asText()).isEqualTo("test_ns.test_table"); assertThat(request.eventType()).isEqualTo(PolarisEventType.AFTER_REFRESH_TABLE.name()); assertThat(request.signature()) .isEqualTo("sha256=" + WebhookEventListener.sign(request.body(), SECRET)); + assertThat( + meterRegistry + .find("polaris.event.webhook.deliveries") + .tag("result", "success") + .counter() + .count()) + .isEqualTo(1.0); } finally { listener.shutdown(); } @@ -176,15 +217,22 @@ void omitsSignatureHeaderWhenNoSecretConfigured() { .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); assertThat(recordedRequests.getFirst().signature()).isNull(); - assertThat(recordedRequests.getFirst().body()).contains(REALM); } finally { listener.shutdown(); } } @Test - void sendsConfiguredCustomHeaders() { - when(config.headers()).thenReturn(Map.of("Authorization", "Splunk test-hec-token")); + void sendsConfiguredCustomHeadersButNotReservedOverrides() { + when(config.headers()) + .thenReturn( + Map.of( + "Authorization", + "Splunk test-hec-token", + WebhookEventListener.SIGNATURE_HEADER, + "sha256=forged", + "Content-Type", + "text/plain")); WebhookEventListener listener = createListener(); try { listener.onEvent(testEvent()); @@ -193,7 +241,11 @@ void sendsConfiguredCustomHeaders() { .atMost(Duration.ofSeconds(10)) .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); - assertThat(recordedRequests.getFirst().authorization()).isEqualTo("Splunk test-hec-token"); + RecordedRequest request = recordedRequests.getFirst(); + assertThat(request.authorization()).isEqualTo("Splunk test-hec-token"); + assertThat(request.signature()) + .isEqualTo("sha256=" + WebhookEventListener.sign(request.body(), SECRET)); + assertThat(request.signature()).doesNotContain("forged"); } finally { listener.shutdown(); } @@ -202,14 +254,25 @@ void sendsConfiguredCustomHeaders() { @Test void failsFastWhenEndpointNotConfigured() { when(config.endpoint()).thenReturn(Optional.empty()); - WebhookEventListener listener = new WebhookEventListener(config, OBJECT_MAPPER, CLOCK); + WebhookEventListener listener = + new WebhookEventListener(config, OBJECT_MAPPER, CLOCK, meterRegistry); assertThatThrownBy(listener::start) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("polaris.event-listener.webhook.endpoint"); } @Test - void retriesFailedDeliveryUntilSuccess() { + void failsFastWhenHttpsRequiredButHttpConfigured() { + when(config.requireHttps()).thenReturn(true); + WebhookEventListener listener = + new WebhookEventListener(config, OBJECT_MAPPER, CLOCK, meterRegistry); + assertThatThrownBy(listener::start) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("https"); + } + + @Test + void retriesTransientFailuresUntilSuccess() { failNext.set(true); WebhookEventListener listener = createListener(); try { @@ -218,6 +281,36 @@ void retriesFailedDeliveryUntilSuccess() { Awaitility.await("delivery retried and succeeded") .atMost(Duration.ofSeconds(10)) .untilAsserted(() -> assertThat(recordedRequests).hasSize(2)); + assertThat(meterRegistry.find("polaris.event.webhook.retries").counter().count()) + .isEqualTo(1.0); + } finally { + listener.shutdown(); + } + } + + @Test + void doesNotRetryPermanentClientErrors() { + statusToReturn.set(400); + when(config.maxAttempts()).thenReturn(5); + WebhookEventListener listener = createListener(); + try { + listener.onEvent(testEvent()); + + Awaitility.await("single attempt only") + .atMost(Duration.ofSeconds(5)) + .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); + // Give a short window to ensure no retry is scheduled. + Awaitility.await().pollDelay(Duration.ofMillis(200)).until(() -> true); + assertThat(recordedRequests).hasSize(1); + assertThat(meterRegistry.find("polaris.event.webhook.retries").counter().count()) + .isEqualTo(0.0); + assertThat( + meterRegistry + .find("polaris.event.webhook.deliveries") + .tag("result", "failure") + .counter() + .count()) + .isEqualTo(1.0); } finally { listener.shutdown(); } @@ -234,12 +327,88 @@ void givesUpAfterMaxAttempts() { Awaitility.await("both attempts performed") .atMost(Duration.ofSeconds(10)) .untilAsserted(() -> assertThat(recordedRequests).hasSize(2)); + } finally { + listener.shutdown(); + } + } - // No further attempts beyond maxAttempts. - assertThat(recordedRequests) - .allSatisfy(r -> assertThat(r.body()).isEqualTo(recordedRequests.getLast().body())); + @Test + void dropsWhenPendingCapacityFull() { + when(config.maxPending()).thenReturn(1); + when(config.maxConcurrent()).thenReturn(1); + // Hold the first delivery so the second is dropped for capacity. + Object gate = new Object(); + AtomicBoolean firstEntered = new AtomicBoolean(false); + server.removeContext("/hook"); + server.createContext( + "/hook", + exchange -> { + firstEntered.set(true); + synchronized (gate) { + try { + gate.wait(5000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + exchange.sendResponseHeaders(200, -1); + exchange.close(); + }); + + WebhookEventListener listener = createListener(); + try { + listener.onEvent(testEvent()); + Awaitility.await("first delivery started") + .atMost(Duration.ofSeconds(5)) + .untilTrue(firstEntered); + + listener.onEvent(testEvent()); + assertThat(meterRegistry.find("polaris.event.webhook.drops").counter().count()) + .isEqualTo(1.0); + + synchronized (gate) { + gate.notifyAll(); + } } finally { + synchronized (gate) { + gate.notifyAll(); + } listener.shutdown(); } } + + @Test + void isTransientClassification() { + assertThat(WebhookEventListener.isTransient(500, null)).isTrue(); + assertThat(WebhookEventListener.isTransient(429, null)).isTrue(); + assertThat(WebhookEventListener.isTransient(408, null)).isTrue(); + assertThat(WebhookEventListener.isTransient(400, null)).isFalse(); + assertThat(WebhookEventListener.isTransient(401, null)).isFalse(); + assertThat(WebhookEventListener.isTransient(404, null)).isFalse(); + assertThat(WebhookEventListener.isTransient(null, new RuntimeException("io"))).isTrue(); + } + + @Test + void parseRetryAfterSeconds() { + HttpHeaders headers = HttpHeaders.of(Map.of("Retry-After", List.of("3")), (a, b) -> true); + assertThat(WebhookEventListener.parseRetryAfterMillis(headers)).contains(3000L); + } + + @Test + void validateEndpointHttps() { + assertThatThrownBy( + () -> + WebhookEventListener.validateEndpoint(URI.create("http://example.com/hook"), true)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("https"); + WebhookEventListener.validateEndpoint(URI.create("https://example.com/hook"), true); + WebhookEventListener.validateEndpoint(URI.create("http://localhost/hook"), false); + } + + @Test + void reservedHeaderDetection() { + assertThat(WebhookEventListener.isReservedHeader("X-Polaris-Signature-256")).isTrue(); + assertThat(WebhookEventListener.isReservedHeader("content-type")).isTrue(); + assertThat(WebhookEventListener.isReservedHeader("Authorization")).isFalse(); + } } diff --git a/runtime/defaults/src/main/resources/application.properties b/runtime/defaults/src/main/resources/application.properties index 1e7eef5f9d..97204a6321 100644 --- a/runtime/defaults/src/main/resources/application.properties +++ b/runtime/defaults/src/main/resources/application.properties @@ -181,6 +181,10 @@ polaris.file-io.type=default # polaris.event-listener.webhook.timeout=10s # polaris.event-listener.webhook.max-attempts=3 # polaris.event-listener.webhook.retry-backoff=1s +# polaris.event-listener.webhook.max-retry-backoff=1m +# polaris.event-listener.webhook.max-concurrent=16 +# polaris.event-listener.webhook.max-pending=1000 +# polaris.event-listener.webhook.require-https=true # OpenTelemetry event listener settings # polaris.event-listener.types=opentelemetry diff --git a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md index cf9fa36e40..bffca1cd0f 100644 --- a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md +++ b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md @@ -23,15 +23,26 @@ build: render: never --- -Configuration interface for the webhook event listener integration. +Configuration for the webhook event listener. -Deliveries are performed asynchronously and do not block the event executor; in-flight requests to a slow or unreachable endpoint are bounded by (`#timeout()`). Delivery is best-effort: retries are kept in memory only, so pending retries are lost on shutdown, and an event that exhausts all attempts is dropped (and logged). Deployments that need stronger delivery guarantees should also enable the `persistence-in-memory-buffer` event listener. HTTP redirects are not followed; the endpoint must respond directly. +Delivery is **best-effort and in-process only**: pending and in-flight deliveries are not durable +and are lost on process restart; events that exhaust retries or cannot be accepted into the bounded +queue are dropped. Ordering is not guaranteed. This listener is not a durable audit spool and is +**not** strengthened by enabling `persistence-in-memory-buffer` (that buffer is a separate listener, +not a webhook delivery queue). + +HTTP redirects are not followed. When `require-https` is `true` (default), only `https` endpoints +are accepted. | Property | Default Value | Type | Description | |----------|---------------|------|-------------| -| `polaris.event-listener.webhook.endpoint` | | `uri` | Returns the HTTP(S) endpoint that webhook events are delivered to.

Each event is sent as an HTTP POST request with a JSON body. The endpoint must return a 2xx status code for the delivery to be considered successful; any other status code (or a connection error) triggers a retry. HTTPS is strongly recommended because the payload may contain sensitive audit data.

This property is required when the `webhook` event listener is enabled.

Configuration property: `polaris.event-listener.webhook.endpoint` | -| `polaris.event-listener.webhook.secret` | | `string` | Returns the shared secret used to sign webhook payloads.

When set, each request carries an `X-Polaris-Signature-256` header containing the hex encoded HMAC-SHA256 of the request body, prefixed with `sha256=`. Receivers should recompute the HMAC over the raw request body and compare it against this header to verify authenticity and integrity. If not set, no signature header is sent.

Configuration property: `polaris.event-listener.webhook.secret` | -| `polaris.event-listener.webhook.headers.`_``_ | | `string` | Returns additional HTTP headers to send with every webhook request.

This is typically used to authenticate against the receiver, for example `polaris.event-listener.webhook.headers."Authorization"=Bearer ` or `polaris.event-listener.webhook.headers."Authorization"=Splunk `. Header values may contain secrets; protect them like any other Polaris credential.

Configuration property: `polaris.event-listener.webhook.headers.""` | -| `polaris.event-listener.webhook.timeout` | `10s` | `duration` | Returns the timeout for a single webhook delivery attempt.

If a delivery attempt does not complete within this duration, it is treated as failed and the retry schedule applies.

Configuration property: `polaris.event-listener.webhook.timeout` | -| `polaris.event-listener.webhook.max-attempts` | `3` | `int` | Returns the maximum number of delivery attempts per event, including the initial attempt.

When all attempts fail, the event is dropped and an error is logged.

Configuration property: `polaris.event-listener.webhook.max-attempts` | -| `polaris.event-listener.webhook.retry-backoff` | `1s` | `duration` | Returns the initial backoff between delivery attempts.

The backoff doubles after each failed attempt (exponential backoff).

Configuration property: `polaris.event-listener.webhook.retry-backoff` | +| `polaris.event-listener.webhook.endpoint` | | `uri` | HTTP(S) endpoint that receives JSON POSTs. Required when the webhook listener is enabled. | +| `polaris.event-listener.webhook.secret` | | `string` | Optional HMAC-SHA256 shared secret (`X-Polaris-Signature-256: sha256=`). Protects authenticity/integrity, not confidentiality. | +| `polaris.event-listener.webhook.headers.`_``_ | | `string` | Extra headers (e.g. `Authorization`). Reserved headers (`Content-Type`, `X-Polaris-Signature-256`, `X-Polaris-Event`, `Host`, `Content-Length`) cannot be overridden. | +| `polaris.event-listener.webhook.timeout` | `10s` | `duration` | Timeout for a single HTTP attempt. | +| `polaris.event-listener.webhook.max-attempts` | `3` | `int` | Max attempts per event including the first. | +| `polaris.event-listener.webhook.retry-backoff` | `1s` | `duration` | Base retry delay (exponential growth with full jitter). | +| `polaris.event-listener.webhook.max-retry-backoff` | `1m` | `duration` | Cap on retry delay (also caps `Retry-After`). | +| `polaris.event-listener.webhook.max-concurrent` | `16` | `int` | Max concurrent HTTP deliveries. | +| `polaris.event-listener.webhook.max-pending` | `1000` | `int` | Max events accepted for delivery/retry; excess are dropped. | +| `polaris.event-listener.webhook.require-https` | `true` | `boolean` | When true, only `https` endpoints are allowed. | diff --git a/site/content/in-dev/unreleased/configuration/configuration-reference.md b/site/content/in-dev/unreleased/configuration/configuration-reference.md index ef79098a51..0e17673106 100644 --- a/site/content/in-dev/unreleased/configuration/configuration-reference.md +++ b/site/content/in-dev/unreleased/configuration/configuration-reference.md @@ -143,12 +143,15 @@ All properties listed here are **runtime** properties and can be changed without {{% include-config-section "smallrye-polaris_event_listener_webhook" %}} The webhook listener delivers each (sanitized) event as an HTTP POST with a JSON body of the -following shape: +following shape (`schema_version` is currently `1`): ```json { + "schema_version": 1, + "event_id": "11111111-2222-3333-4444-555555555555", "event_type": "AFTER_REFRESH_TABLE", "timestamp": 1760000000000, + "delivery_timestamp": 1760000000500, "realm_id": "my-realm", "principal": "alice", "activated_roles": ["role1", "role2"], @@ -157,16 +160,22 @@ following shape: } ``` -`principal`, `activated_roles`, `request_id`, and `table_identifier` are omitted when not -applicable to the event. Each request also carries an `X-Polaris-Event` header with the event +`timestamp` is the original event time from Polaris event metadata; `delivery_timestamp` is when +the payload was serialized for send. `event_id` is stable for the event and can be used by receivers +to detect retry duplicates. `principal`, `activated_roles`, `request_id`, and `table_identifier` +are omitted when not applicable. Each request carries an `X-Polaris-Event` header with the event type. When `polaris.event-listener.webhook.secret` is set, an `X-Polaris-Signature-256` header -contains `sha256=` followed by the hex-encoded HMAC-SHA256 of the raw request body; receivers -should recompute the HMAC with the shared secret and compare. - -Delivery is best-effort: failed deliveries are retried with exponential backoff and dropped after -`max-attempts`. Retries are in-memory only, so pending retries are lost on restart. Deployments -that need stronger delivery guarantees should also enable the `persistence-in-memory-buffer` -listener. Redirects are not followed; the endpoint must respond directly with a 2xx status code. +contains `sha256=` followed by the hex-encoded HMAC-SHA256 of the raw request body. + +Delivery is **best-effort and bounded**: concurrency and pending work are capped +(`max-concurrent`, `max-pending`); excess events are dropped. Only transient failures (network +errors, 408/429/5xx) are retried, with exponential backoff, full jitter, optional `Retry-After`, +and a max backoff. Permanent 4xx responses are not retried. Retries and in-flight work are +in-memory only and lost on restart. This is **not** a durable spool; enabling +`persistence-in-memory-buffer` does not replay webhook deliveries. By default only HTTPS endpoints +are accepted (`require-https=true`). Redirects are not followed. Metrics include +`polaris.event.webhook.deliveries`, `retries`, `drops`, `pending`, `in_flight`, and `delivery` +latency. ### `opentelemetry` event listener From 4dc85dc2af0397a378a1dc3e5ef19ea872e33adc Mon Sep 17 00:00:00 2001 From: vignesh_A Date: Tue, 21 Jul 2026 02:55:22 +0530 Subject: [PATCH 3/5] refactor(events): CloudEvents-shaped webhook envelope, drop redundant semaphore --- CHANGELOG.md | 2 +- .../events/webhook/WebhookEventListener.java | 34 ++++++++----------- .../webhook/WebhookEventListenerTest.java | 12 +++---- .../configuration/configuration-reference.md | 29 +++++++++------- 4 files changed, 37 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 78689e14c4..4adbf56465 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,7 +48,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti - Added optional `sessionPolicy` field to `SigV4AuthenticationParameters` for catalog federation. When set, the IAM session policy JSON is attached to the STS AssumeRole request, allowing administrators to restrict vended credentials to only the required AWS services and actions (Principle of Least Privilege). - Python CLI: added `--scheme` to specify URL scheme - Added opt-in idempotency for `createTable` and `updateTable` in the Iceberg REST catalog. When enabled via `polaris.idempotency.enabled=true` (default `false`), a client-supplied `Idempotency-Key` header is embedded into the table entity and committed in the same transaction as the operation; a retry carrying the same key within the TTL window (`polaris.idempotency.ttl`, default `PT5M`) replays the original success instead of failing — with `AlreadyExists` for `createTable`, or with `CommitFailedException` for `updateTable` when the request's requirements no longer match the already-advanced table. -- Added a `webhook` event listener (`polaris.event-listener.types=webhook`) that delivers sanitized Polaris events as JSON HTTP POSTs to a configurable endpoint (SIEM/audit collectors). Payloads are a versioned envelope (`schema_version`, `event_id`, original `timestamp`, optional `delivery_timestamp`) with optional HMAC-SHA256 signing (`X-Polaris-Signature-256`). Delivery is best-effort and bounded (`max-concurrent`, `max-pending`); transient failures retry with jitter (and `Retry-After` when present); HTTPS is required by default (`require-https`); reserved signature/event headers cannot be overridden by custom headers. Metrics cover success, failure, retries, drops, pending, in-flight, and latency. +- Added a `webhook` event listener (`polaris.event-listener.types=webhook`) that delivers sanitized Polaris events as JSON HTTP POSTs to a configurable endpoint (SIEM/audit collectors). Payloads use a CloudEvents-shaped envelope (`specversion`, `id`, `type`, `source`, original `time`, plus `delivery_time`) with optional HMAC-SHA256 signing (`X-Polaris-Signature-256`). Delivery is best-effort and bounded (`max-concurrent`, `max-pending`); transient failures retry with jitter (and `Retry-After` when present); HTTPS is required by default (`require-https`); reserved signature/event headers cannot be overridden by custom headers. Metrics cover success, failure, retries, drops, pending, in-flight, and latency. ### Changes - The admin tool's `bootstrap` command is now idempotent: bootstrapping a realm that is already diff --git a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java index be882b0626..52f0bc6ed5 100644 --- a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java @@ -49,7 +49,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import javax.crypto.Mac; @@ -75,7 +74,12 @@ public class WebhookEventListener implements PolarisEventListener { static final String SIGNATURE_HEADER = "X-Polaris-Signature-256"; static final String EVENT_TYPE_HEADER = "X-Polaris-Event"; - static final int SCHEMA_VERSION = 1; + + /** CloudEvents spec version the payload envelope is shaped after. */ + static final String SPEC_VERSION = "1.0"; + + /** CloudEvents {@code source} attribute identifying Polaris as the event producer. */ + static final String EVENT_SOURCE = "org.apache.polaris"; private static final Set RESERVED_HEADERS = Set.of( @@ -106,7 +110,6 @@ public class WebhookEventListener implements PolarisEventListener { private HttpClient client; private ExecutorService deliveryExecutor; private ScheduledExecutorService retryExecutor; - private Semaphore concurrency; private Counter successCounter; private Counter failureCounter; @@ -144,7 +147,6 @@ void start() { } validateEndpoint(endpoint, requireHttps); this.client = createHttpClient(); - this.concurrency = new Semaphore(maxConcurrent); this.deliveryExecutor = Executors.newFixedThreadPool( maxConcurrent, @@ -322,15 +324,6 @@ private void scheduleDelivery(String payload, String eventType, int attempt, lon } private void deliverWithConcurrency(String payload, String eventType, int attempt) { - try { - concurrency.acquire(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - releaseAccepted(); - dropCounter.increment(); - LOGGER.warn("Interrupted while waiting for webhook concurrency; dropping event"); - return; - } inFlight.incrementAndGet(); long startNanos = System.nanoTime(); try { @@ -353,7 +346,6 @@ private void deliverWithConcurrency(String payload, String eventType, int attemp handleFailure(payload, eventType, attempt, null, null, e); } finally { inFlight.decrementAndGet(); - concurrency.release(); } } @@ -472,12 +464,14 @@ static boolean isReservedHeader(String name) { @VisibleForTesting String toJson(PolarisEvent event) { HashMap properties = new HashMap<>(); - properties.put("schema_version", SCHEMA_VERSION); - properties.put("event_id", event.metadata().eventId().toString()); - properties.put("event_type", event.type().name()); - // Original event time from metadata (not delivery time). - properties.put("timestamp", event.metadata().timestamp().toEpochMilli()); - properties.put("delivery_timestamp", clock.millis()); + // CloudEvents-shaped envelope (no SDK): stable id, original event time, spec version. + properties.put("specversion", SPEC_VERSION); + properties.put("id", event.metadata().eventId().toString()); + properties.put("type", event.type().name()); + properties.put("source", EVENT_SOURCE); + // Original event time from metadata (not delivery time), RFC 3339 per CloudEvents. + properties.put("time", event.metadata().timestamp().toString()); + properties.put("delivery_time", clock.instant().toString()); event .attributes() .get(EventAttributes.TABLE_IDENTIFIER) diff --git a/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java index 5d4e1cc38b..d873f97bdc 100644 --- a/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java +++ b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java @@ -181,12 +181,12 @@ void deliversSignedJsonPayloadWithEnvelopeFields() throws Exception { RecordedRequest request = recordedRequests.getFirst(); JsonNode json = OBJECT_MAPPER.readTree(request.body()); - assertThat(json.get("schema_version").asInt()).isEqualTo(WebhookEventListener.SCHEMA_VERSION); - assertThat(json.get("event_id").asText()).isEqualTo(EVENT_ID.toString()); - assertThat(json.get("timestamp").asLong()).isEqualTo(EVENT_TIME.toEpochMilli()); - assertThat(json.get("delivery_timestamp").asLong()).isPositive(); - assertThat(json.get("event_type").asText()) - .isEqualTo(PolarisEventType.AFTER_REFRESH_TABLE.name()); + assertThat(json.get("specversion").asText()).isEqualTo(WebhookEventListener.SPEC_VERSION); + assertThat(json.get("id").asText()).isEqualTo(EVENT_ID.toString()); + assertThat(json.get("time").asText()).isEqualTo(EVENT_TIME.toString()); + assertThat(json.get("delivery_time").asText()).isNotBlank(); + assertThat(json.get("source").asText()).isEqualTo(WebhookEventListener.EVENT_SOURCE); + assertThat(json.get("type").asText()).isEqualTo(PolarisEventType.AFTER_REFRESH_TABLE.name()); assertThat(json.get("realm_id").asText()).isEqualTo(REALM); assertThat(json.get("principal").asText()).isEqualTo(TEST_USER); assertThat(json.get("table_identifier").asText()).isEqualTo("test_ns.test_table"); diff --git a/site/content/in-dev/unreleased/configuration/configuration-reference.md b/site/content/in-dev/unreleased/configuration/configuration-reference.md index 0e17673106..222008871f 100644 --- a/site/content/in-dev/unreleased/configuration/configuration-reference.md +++ b/site/content/in-dev/unreleased/configuration/configuration-reference.md @@ -142,16 +142,17 @@ All properties listed here are **runtime** properties and can be changed without {{% include-config-section "smallrye-polaris_event_listener_webhook" %}} -The webhook listener delivers each (sanitized) event as an HTTP POST with a JSON body of the -following shape (`schema_version` is currently `1`): +The webhook listener delivers each (sanitized) event as an HTTP POST with a JSON body shaped after +the CloudEvents envelope (no CloudEvents SDK required): ```json { - "schema_version": 1, - "event_id": "11111111-2222-3333-4444-555555555555", - "event_type": "AFTER_REFRESH_TABLE", - "timestamp": 1760000000000, - "delivery_timestamp": 1760000000500, + "specversion": "1.0", + "id": "11111111-2222-3333-4444-555555555555", + "type": "AFTER_REFRESH_TABLE", + "source": "org.apache.polaris", + "time": "2026-01-15T12:00:00Z", + "delivery_time": "2026-01-15T12:00:01Z", "realm_id": "my-realm", "principal": "alice", "activated_roles": ["role1", "role2"], @@ -160,8 +161,8 @@ following shape (`schema_version` is currently `1`): } ``` -`timestamp` is the original event time from Polaris event metadata; `delivery_timestamp` is when -the payload was serialized for send. `event_id` is stable for the event and can be used by receivers +`time` is the original event time from Polaris event metadata (RFC 3339); `delivery_time` is when +the payload was serialized for send. `id` is stable for the event and can be used by receivers to detect retry duplicates. `principal`, `activated_roles`, `request_id`, and `table_identifier` are omitted when not applicable. Each request carries an `X-Polaris-Event` header with the event type. When `polaris.event-listener.webhook.secret` is set, an `X-Polaris-Signature-256` header @@ -169,10 +170,12 @@ contains `sha256=` followed by the hex-encoded HMAC-SHA256 of the raw request bo Delivery is **best-effort and bounded**: concurrency and pending work are capped (`max-concurrent`, `max-pending`); excess events are dropped. Only transient failures (network -errors, 408/429/5xx) are retried, with exponential backoff, full jitter, optional `Retry-After`, -and a max backoff. Permanent 4xx responses are not retried. Retries and in-flight work are -in-memory only and lost on restart. This is **not** a durable spool; enabling -`persistence-in-memory-buffer` does not replay webhook deliveries. By default only HTTPS endpoints +errors, 408/425/429/5xx) are retried, with exponential backoff, full jitter, honoring +`Retry-After` (delta-seconds), and a max backoff. Permanent 4xx responses are not retried. Retries +and in-flight work are in-memory only and lost on restart. This is **not** a durable spool; enabling +`persistence-in-memory-buffer` does not replay webhook deliveries. Deployments that need durable, +replayable delivery should instead consume Polaris events from Kafka and run their own webhook +processor with endpoint-specific retry and dead-letter handling. By default only HTTPS endpoints are accepted (`require-https=true`). Redirects are not followed. Metrics include `polaris.event.webhook.deliveries`, `retries`, `drops`, `pending`, `in_flight`, and `delivery` latency. From dd7c5daaa87adbd51298a30b25f880cc169628cc Mon Sep 17 00:00:00 2001 From: vignesh_A Date: Tue, 21 Jul 2026 03:07:07 +0530 Subject: [PATCH 4/5] refactor(events): compliant CloudEvents attribute names, full Retry-After, deterministic field order --- CHANGELOG.md | 2 +- .../events/webhook/WebhookEventListener.java | 40 ++++++++++++++----- .../webhook/WebhookEventListenerTest.java | 28 +++++++++++-- .../configuration/configuration-reference.md | 18 +++++---- 4 files changed, 64 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4adbf56465..ae09a5cf63 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,7 +48,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti - Added optional `sessionPolicy` field to `SigV4AuthenticationParameters` for catalog federation. When set, the IAM session policy JSON is attached to the STS AssumeRole request, allowing administrators to restrict vended credentials to only the required AWS services and actions (Principle of Least Privilege). - Python CLI: added `--scheme` to specify URL scheme - Added opt-in idempotency for `createTable` and `updateTable` in the Iceberg REST catalog. When enabled via `polaris.idempotency.enabled=true` (default `false`), a client-supplied `Idempotency-Key` header is embedded into the table entity and committed in the same transaction as the operation; a retry carrying the same key within the TTL window (`polaris.idempotency.ttl`, default `PT5M`) replays the original success instead of failing — with `AlreadyExists` for `createTable`, or with `CommitFailedException` for `updateTable` when the request's requirements no longer match the already-advanced table. -- Added a `webhook` event listener (`polaris.event-listener.types=webhook`) that delivers sanitized Polaris events as JSON HTTP POSTs to a configurable endpoint (SIEM/audit collectors). Payloads use a CloudEvents-shaped envelope (`specversion`, `id`, `type`, `source`, original `time`, plus `delivery_time`) with optional HMAC-SHA256 signing (`X-Polaris-Signature-256`). Delivery is best-effort and bounded (`max-concurrent`, `max-pending`); transient failures retry with jitter (and `Retry-After` when present); HTTPS is required by default (`require-https`); reserved signature/event headers cannot be overridden by custom headers. Metrics cover success, failure, retries, drops, pending, in-flight, and latency. +- Added a `webhook` event listener (`polaris.event-listener.types=webhook`) that delivers sanitized Polaris events as JSON HTTP POSTs to a configurable endpoint (SIEM/audit collectors). Payloads use a CloudEvents-shaped envelope (`specversion`, `id`, `type`, `source`, original `time`, plus `deliverytime` and CloudEvents-compliant extension attribute names) with optional HMAC-SHA256 signing (`X-Polaris-Signature-256`). Delivery is best-effort and bounded (`max-concurrent`, `max-pending`); transient failures retry with jitter (and `Retry-After` when present); HTTPS is required by default (`require-https`); reserved signature/event headers cannot be overridden by custom headers. Metrics cover success, failure, retries, drops, pending, in-flight, and latency. ### Changes - The admin tool's `bootstrap` command is now idempotent: bootstrapping a realm that is already diff --git a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java index 52f0bc6ed5..40a527d994 100644 --- a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java @@ -41,7 +41,10 @@ import java.security.SecureRandom; import java.time.Clock; import java.time.Duration; -import java.util.HashMap; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; +import java.util.LinkedHashMap; import java.util.Locale; import java.util.Map; import java.util.Optional; @@ -399,7 +402,7 @@ static boolean isTransient(Integer status, Throwable error) { @VisibleForTesting long computeRetryDelayMillis(int attempt, HttpHeaders headers) { - Optional retryAfter = parseRetryAfterMillis(headers); + Optional retryAfter = parseRetryAfterMillis(headers, clock); if (retryAfter.isPresent()) { return Math.min(retryAfter.get(), maxRetryBackoff.toMillis()); } @@ -412,8 +415,13 @@ long computeRetryDelayMillis(int attempt, HttpHeaders headers) { return (long) (random.nextDouble() * capped); } + /** + * Parses the {@code Retry-After} header in either of its two standard forms: delta-seconds or + * HTTP-date (RFC 9110). Returns the requested delay in milliseconds, clamped at zero for dates in + * the past, or empty when the header is absent or malformed. + */ @VisibleForTesting - static Optional parseRetryAfterMillis(HttpHeaders headers) { + static Optional parseRetryAfterMillis(HttpHeaders headers, Clock clock) { if (headers == null) { return Optional.empty(); } @@ -423,13 +431,22 @@ static Optional parseRetryAfterMillis(HttpHeaders headers) { .flatMap( value -> { try { - // Delta-seconds form only (HTTP-date not supported). long seconds = Long.parseLong(value); if (seconds < 0) { return Optional.empty(); } return Optional.of(TimeUnit.SECONDS.toMillis(seconds)); } catch (NumberFormatException e) { + // Not delta-seconds; try HTTP-date below. + } + try { + long delayMs = + ZonedDateTime.parse(value, DateTimeFormatter.RFC_1123_DATE_TIME) + .toInstant() + .toEpochMilli() + - clock.millis(); + return Optional.of(Math.max(delayMs, 0L)); + } catch (DateTimeParseException e) { return Optional.empty(); } }); @@ -463,30 +480,31 @@ static boolean isReservedHeader(String name) { @VisibleForTesting String toJson(PolarisEvent event) { - HashMap properties = new HashMap<>(); - // CloudEvents-shaped envelope (no SDK): stable id, original event time, spec version. + // LinkedHashMap for deterministic field order: core CloudEvents attributes first, then + // extension attributes (compliant names: lowercase alphanumerics). + Map properties = new LinkedHashMap<>(); properties.put("specversion", SPEC_VERSION); properties.put("id", event.metadata().eventId().toString()); properties.put("type", event.type().name()); properties.put("source", EVENT_SOURCE); // Original event time from metadata (not delivery time), RFC 3339 per CloudEvents. properties.put("time", event.metadata().timestamp().toString()); - properties.put("delivery_time", clock.instant().toString()); + properties.put("deliverytime", clock.instant().toString()); + properties.put("realmid", event.metadata().realmId()); event .attributes() .get(EventAttributes.TABLE_IDENTIFIER) .map(TableIdentifier::toString) - .ifPresent(id -> properties.put("table_identifier", id)); - properties.put("realm_id", event.metadata().realmId()); + .ifPresent(id -> properties.put("tableidentifier", id)); event .metadata() .user() .ifPresent( p -> { properties.put("principal", p.getName()); - properties.put("activated_roles", p.getRoles()); + properties.put("activatedroles", p.getRoles()); }); - event.metadata().requestId().ifPresent(id -> properties.put("request_id", id)); + event.metadata().requestId().ifPresent(id -> properties.put("requestid", id)); try { return objectMapper.writeValueAsString(properties); } catch (JsonProcessingException e) { diff --git a/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java index d873f97bdc..de6c86acc1 100644 --- a/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java +++ b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java @@ -35,6 +35,7 @@ import java.time.Clock; import java.time.Duration; import java.time.Instant; +import java.time.ZoneOffset; import java.util.List; import java.util.Map; import java.util.Optional; @@ -184,12 +185,13 @@ void deliversSignedJsonPayloadWithEnvelopeFields() throws Exception { assertThat(json.get("specversion").asText()).isEqualTo(WebhookEventListener.SPEC_VERSION); assertThat(json.get("id").asText()).isEqualTo(EVENT_ID.toString()); assertThat(json.get("time").asText()).isEqualTo(EVENT_TIME.toString()); - assertThat(json.get("delivery_time").asText()).isNotBlank(); + assertThat(json.get("deliverytime").asText()).isNotBlank(); assertThat(json.get("source").asText()).isEqualTo(WebhookEventListener.EVENT_SOURCE); assertThat(json.get("type").asText()).isEqualTo(PolarisEventType.AFTER_REFRESH_TABLE.name()); - assertThat(json.get("realm_id").asText()).isEqualTo(REALM); + assertThat(json.get("realmid").asText()).isEqualTo(REALM); assertThat(json.get("principal").asText()).isEqualTo(TEST_USER); - assertThat(json.get("table_identifier").asText()).isEqualTo("test_ns.test_table"); + assertThat(json.get("activatedroles")).hasSize(2); + assertThat(json.get("tableidentifier").asText()).isEqualTo("test_ns.test_table"); assertThat(request.eventType()).isEqualTo(PolarisEventType.AFTER_REFRESH_TABLE.name()); assertThat(request.signature()) .isEqualTo("sha256=" + WebhookEventListener.sign(request.body(), SECRET)); @@ -391,7 +393,25 @@ void isTransientClassification() { @Test void parseRetryAfterSeconds() { HttpHeaders headers = HttpHeaders.of(Map.of("Retry-After", List.of("3")), (a, b) -> true); - assertThat(WebhookEventListener.parseRetryAfterMillis(headers)).contains(3000L); + assertThat(WebhookEventListener.parseRetryAfterMillis(headers, CLOCK)).contains(3000L); + } + + @Test + void parseRetryAfterHttpDate() { + Clock fixed = Clock.fixed(EVENT_TIME, ZoneOffset.UTC); + HttpHeaders future = + HttpHeaders.of( + Map.of("Retry-After", List.of("Thu, 15 Jan 2026 12:00:30 GMT")), (a, b) -> true); + assertThat(WebhookEventListener.parseRetryAfterMillis(future, fixed)).contains(30000L); + // Dates in the past clamp to zero delay. + HttpHeaders past = + HttpHeaders.of( + Map.of("Retry-After", List.of("Thu, 15 Jan 2026 11:59:00 GMT")), (a, b) -> true); + assertThat(WebhookEventListener.parseRetryAfterMillis(past, fixed)).contains(0L); + // Malformed values are ignored so computed backoff applies. + HttpHeaders malformed = + HttpHeaders.of(Map.of("Retry-After", List.of("not-a-date")), (a, b) -> true); + assertThat(WebhookEventListener.parseRetryAfterMillis(malformed, fixed)).isEmpty(); } @Test diff --git a/site/content/in-dev/unreleased/configuration/configuration-reference.md b/site/content/in-dev/unreleased/configuration/configuration-reference.md index 222008871f..228105842b 100644 --- a/site/content/in-dev/unreleased/configuration/configuration-reference.md +++ b/site/content/in-dev/unreleased/configuration/configuration-reference.md @@ -152,18 +152,19 @@ the CloudEvents envelope (no CloudEvents SDK required): "type": "AFTER_REFRESH_TABLE", "source": "org.apache.polaris", "time": "2026-01-15T12:00:00Z", - "delivery_time": "2026-01-15T12:00:01Z", - "realm_id": "my-realm", + "deliverytime": "2026-01-15T12:00:01Z", + "realmid": "my-realm", "principal": "alice", - "activated_roles": ["role1", "role2"], - "request_id": "d3b0...", - "table_identifier": "ns.tbl" + "activatedroles": ["role1", "role2"], + "requestid": "d3b0...", + "tableidentifier": "ns.tbl" } ``` -`time` is the original event time from Polaris event metadata (RFC 3339); `delivery_time` is when +`time` is the original event time from Polaris event metadata (RFC 3339); `deliverytime` is when the payload was serialized for send. `id` is stable for the event and can be used by receivers -to detect retry duplicates. `principal`, `activated_roles`, `request_id`, and `table_identifier` +to detect retry duplicates. Extension attributes use CloudEvents-compliant names (lowercase +alphanumerics). `principal`, `activatedroles`, `requestid`, and `tableidentifier` are omitted when not applicable. Each request carries an `X-Polaris-Event` header with the event type. When `polaris.event-listener.webhook.secret` is set, an `X-Polaris-Signature-256` header contains `sha256=` followed by the hex-encoded HMAC-SHA256 of the raw request body. @@ -171,7 +172,8 @@ contains `sha256=` followed by the hex-encoded HMAC-SHA256 of the raw request bo Delivery is **best-effort and bounded**: concurrency and pending work are capped (`max-concurrent`, `max-pending`); excess events are dropped. Only transient failures (network errors, 408/425/429/5xx) are retried, with exponential backoff, full jitter, honoring -`Retry-After` (delta-seconds), and a max backoff. Permanent 4xx responses are not retried. Retries +`Retry-After` (delta-seconds or HTTP-date), and a max backoff. Permanent 4xx responses are not +retried. Retries and in-flight work are in-memory only and lost on restart. This is **not** a durable spool; enabling `persistence-in-memory-buffer` does not replay webhook deliveries. Deployments that need durable, replayable delivery should instead consume Polaris events from Kafka and run their own webhook From 30c48869d9764348d2f1144f76b98bcc9c379d89 Mon Sep 17 00:00:00 2001 From: vignesh_A Date: Tue, 21 Jul 2026 03:55:42 +0530 Subject: [PATCH 5/5] docs(events): regenerate webhook config reference from javadoc --- .../WebhookEventListenerConfiguration.java | 8 ++--- ...smallrye-polaris_event_listener_webhook.md | 31 ++++++++----------- 2 files changed, 17 insertions(+), 22 deletions(-) diff --git a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java index 3b862dcb6f..8954154dbd 100644 --- a/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java @@ -37,8 +37,8 @@ * durable audit spool and is not strengthened by enabling {@code persistence-in-memory-buffer} * (that buffer is a separate listener, not a webhook delivery queue). * - *

HTTP redirects are not followed; the endpoint must respond directly. When {@link - * #requireHttps()} is {@code true} (the default), only {@code https} endpoints are accepted. + *

HTTP redirects are not followed; the endpoint must respond directly. When {@code + * require-https} is {@code true} (the default), only {@code https} endpoints are accepted. */ @StaticInitSafe @ConfigMapping(prefix = "polaris.event-listener.webhook") @@ -98,7 +98,7 @@ public interface WebhookEventListenerConfiguration { /** * Base delay before the first retry. Subsequent delays grow exponentially with full jitter, - * capped by {@link #maxRetryBackoff()}. + * capped by {@code max-retry-backoff}. * *

Configuration property: {@code polaris.event-listener.webhook.retry-backoff} */ @@ -117,7 +117,7 @@ public interface WebhookEventListenerConfiguration { /** * Maximum concurrent HTTP deliveries (including retries currently executing). Additional work - * waits on the concurrency limit while still counting toward {@link #maxPending()}. + * waits on the concurrency limit while still counting toward {@code max-pending}. * *

Configuration property: {@code polaris.event-listener.webhook.max-concurrent} */ diff --git a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md index bffca1cd0f..ad79d64cc6 100644 --- a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md +++ b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md @@ -23,26 +23,21 @@ build: render: never --- -Configuration for the webhook event listener. +Configuration for the webhook event listener. -Delivery is **best-effort and in-process only**: pending and in-flight deliveries are not durable -and are lost on process restart; events that exhaust retries or cannot be accepted into the bounded -queue are dropped. Ordering is not guaranteed. This listener is not a durable audit spool and is -**not** strengthened by enabling `persistence-in-memory-buffer` (that buffer is a separate listener, -not a webhook delivery queue). +Delivery is best-effort and in-process only: pending and in-flight deliveries are not durable and are lost on process restart; events that exhaust retries or cannot be accepted into the bounded queue are dropped. Ordering is not guaranteed. This listener is not a durable audit spool and is not strengthened by enabling `persistence-in-memory-buffer` (that buffer is a separate listener, not a webhook delivery queue). -HTTP redirects are not followed. When `require-https` is `true` (default), only `https` endpoints -are accepted. +HTTP redirects are not followed; the endpoint must respond directly. When `require-https` is `true` (the default), only `https` endpoints are accepted. | Property | Default Value | Type | Description | |----------|---------------|------|-------------| -| `polaris.event-listener.webhook.endpoint` | | `uri` | HTTP(S) endpoint that receives JSON POSTs. Required when the webhook listener is enabled. | -| `polaris.event-listener.webhook.secret` | | `string` | Optional HMAC-SHA256 shared secret (`X-Polaris-Signature-256: sha256=`). Protects authenticity/integrity, not confidentiality. | -| `polaris.event-listener.webhook.headers.`_``_ | | `string` | Extra headers (e.g. `Authorization`). Reserved headers (`Content-Type`, `X-Polaris-Signature-256`, `X-Polaris-Event`, `Host`, `Content-Length`) cannot be overridden. | -| `polaris.event-listener.webhook.timeout` | `10s` | `duration` | Timeout for a single HTTP attempt. | -| `polaris.event-listener.webhook.max-attempts` | `3` | `int` | Max attempts per event including the first. | -| `polaris.event-listener.webhook.retry-backoff` | `1s` | `duration` | Base retry delay (exponential growth with full jitter). | -| `polaris.event-listener.webhook.max-retry-backoff` | `1m` | `duration` | Cap on retry delay (also caps `Retry-After`). | -| `polaris.event-listener.webhook.max-concurrent` | `16` | `int` | Max concurrent HTTP deliveries. | -| `polaris.event-listener.webhook.max-pending` | `1000` | `int` | Max events accepted for delivery/retry; excess are dropped. | -| `polaris.event-listener.webhook.require-https` | `true` | `boolean` | When true, only `https` endpoints are allowed. | +| `polaris.event-listener.webhook.endpoint` | | `uri` | HTTP(S) endpoint that receives webhook POSTs with a JSON body.

A 2xx response marks success. Transient failures (network errors, 408/429/5xx) are retried within the concurrency limits; permanent client errors (other 4xx) are not retried.

Required when the `webhook` event listener is enabled.

Configuration property: `polaris.event-listener.webhook.endpoint` | +| `polaris.event-listener.webhook.secret` | | `string` | Shared secret for HMAC-SHA256 signing. When set, each request includes `X-Polaris-Signature-256: sha256=` over the raw body. HMAC authenticates and protects integrity; it does not encrypt the payload — use HTTPS for confidentiality.

Configuration property: `polaris.event-listener.webhook.secret` | +| `polaris.event-listener.webhook.headers.`_``_ | | `string` | Extra headers on every request (for example `Authorization`). Reserved headers ( `Content-Type` , `X-Polaris-Signature-256`, `X-Polaris-Event`, `Host`, `Content-Length` ) cannot be overridden. Header values may be secrets; protect them accordingly.

Configuration property: `polaris.event-listener.webhook.headers.""` | +| `polaris.event-listener.webhook.timeout` | `10s` | `duration` | Timeout for a single HTTP attempt. Timed-out attempts count as transient failures.

Configuration property: `polaris.event-listener.webhook.timeout` | +| `polaris.event-listener.webhook.max-attempts` | `3` | `int` | Maximum delivery attempts per event including the first try. After this, the event is dropped.

Configuration property: `polaris.event-listener.webhook.max-attempts` | +| `polaris.event-listener.webhook.retry-backoff` | `1s` | `duration` | Base delay before the first retry. Subsequent delays grow exponentially with full jitter, capped by `max-retry-backoff`.

Configuration property: `polaris.event-listener.webhook.retry-backoff` | +| `polaris.event-listener.webhook.max-retry-backoff` | `1m` | `duration` | Upper bound on retry delay after exponential growth and jitter.

Configuration property: `polaris.event-listener.webhook.max-retry-backoff` | +| `polaris.event-listener.webhook.max-concurrent` | `16` | `int` | Maximum concurrent HTTP deliveries (including retries currently executing). Additional work waits on the concurrency limit while still counting toward `max-pending`.

Configuration property: `polaris.event-listener.webhook.max-concurrent` | +| `polaris.event-listener.webhook.max-pending` | `1000` | `int` | Maximum events accepted for delivery or retry at once. When full, new events are dropped.

Configuration property: `polaris.event-listener.webhook.max-pending` | +| `polaris.event-listener.webhook.require-https` | `true` | `boolean` | When `true`, only `https` endpoints are allowed (recommended for production). Set `false` only for local testing with plain HTTP.

Configuration property: `polaris.event-listener.webhook.require-https` |