diff --git a/CHANGELOG.md b/CHANGELOG.md index ec7a853939..ae09a5cf63 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 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/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..0ed45dbc04 --- /dev/null +++ b/extensions/events/webhook/build.gradle.kts @@ -0,0 +1,52 @@ +/* + * 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") + implementation("io.micrometer:micrometer-core") + implementation(libs.guava) + + 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..40a527d994 --- /dev/null +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListener.java @@ -0,0 +1,542 @@ +/* + * 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 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; +import jakarta.enterprise.context.ApplicationScoped; +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.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; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +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; +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 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") +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"; + + /** 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( + "content-type", + "content-length", + "host", + SIGNATURE_HEADER.toLowerCase(Locale.ROOT), + EVENT_TYPE_HEADER.toLowerCase(Locale.ROOT)); + + 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 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 Counter successCounter; + private Counter failureCounter; + private Counter retryCounter; + private Counter dropCounter; + private Timer deliveryTimer; + + @Inject + public WebhookEventListener( + WebhookEventListenerConfiguration config, + ObjectMapper objectMapper, + Clock clock, + MeterRegistry meterRegistry) { + this.endpoint = config.endpoint().orElse(null); + this.secret = config.secret().orElse(null); + 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 + void start() { + if (endpoint == null) { + throw new IllegalStateException( + "The webhook event listener is enabled but polaris.event-listener.webhook.endpoint is" + + " not set"); + } + validateEndpoint(endpoint, requireHttps); + this.client = createHttpClient(); + 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 -> { + Thread thread = new Thread(runnable, "polaris-webhook-event-listener-retry"); + 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) + .followRedirects(HttpClient.Redirect.NEVER) + .build(); + } + + @PreDestroy + void shutdown() { + if (retryExecutor != null) { + 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) { + 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(); + } + } + + 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; + } + 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, clock); + 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); + } + + /** + * 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, Clock clock) { + if (headers == null) { + return Optional.empty(); + } + return headers + .firstValue("Retry-After") + .map(String::trim) + .flatMap( + value -> { + try { + 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(); + } + }); + } + + 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) + .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)); + } + return requestBuilder.build(); + } + + @VisibleForTesting + static boolean isReservedHeader(String name) { + return RESERVED_HEADERS.contains(name.toLowerCase(Locale.ROOT)); + } + + @VisibleForTesting + String toJson(PolarisEvent event) { + // 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("deliverytime", clock.instant().toString()); + properties.put("realmid", event.metadata().realmId()); + event + .attributes() + .get(EventAttributes.TABLE_IDENTIFIER) + .map(TableIdentifier::toString) + .ifPresent(id -> properties.put("tableidentifier", id)); + event + .metadata() + .user() + .ifPresent( + p -> { + properties.put("principal", p.getName()); + properties.put("activatedroles", p.getRoles()); + }); + event.metadata().requestId().ifPresent(id -> properties.put("requestid", 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); + } + } + + @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 new file mode 100644 index 0000000000..8954154dbd --- /dev/null +++ b/extensions/events/webhook/src/main/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerConfiguration.java @@ -0,0 +1,146 @@ +/* + * 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 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 {@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 {@code + * require-https} is {@code true} (the default), only {@code https} endpoints are accepted. + */ +@StaticInitSafe +@ConfigMapping(prefix = "polaris.event-listener.webhook") +@ApplicationScoped +public interface WebhookEventListenerConfiguration { + + /** + * 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 {@code webhook} event listener is enabled. + * + *

Configuration property: {@code polaris.event-listener.webhook.endpoint} + */ + @WithName("endpoint") + Optional endpoint(); + + /** + * 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} + */ + @WithName("secret") + Optional secret(); + + /** + * 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.""} + */ + @WithName("headers") + Map headers(); + + /** + * Timeout for a single HTTP attempt. Timed-out attempts count as transient failures. + * + *

Configuration property: {@code polaris.event-listener.webhook.timeout} + */ + @WithName("timeout") + @WithDefault("10s") + Duration timeout(); + + /** + * Maximum delivery attempts per event including the first try. After this, the event is dropped. + * + *

Configuration property: {@code polaris.event-listener.webhook.max-attempts} + */ + @WithName("max-attempts") + @WithDefault("3") + int maxAttempts(); + + /** + * Base delay before the first retry. Subsequent delays grow exponentially with full jitter, + * capped by {@code max-retry-backoff}. + * + *

Configuration property: {@code polaris.event-listener.webhook.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 {@code max-pending}. + * + *

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 new file mode 100644 index 0000000000..de6c86acc1 --- /dev/null +++ b/extensions/events/webhook/src/test/java/org/apache/polaris/extensions/events/webhook/WebhookEventListenerTest.java @@ -0,0 +1,434 @@ +/* + * 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.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.time.ZoneOffset; +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; +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 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(); + + 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 AtomicInteger statusToReturn; + private SimpleMeterRegistry meterRegistry; + + 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); + statusToReturn = new AtomicInteger(200); + meterRegistry = new SimpleMeterRegistry(); + 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; + if (failAlways.get()) { + status = 500; + } else if (failNext.getAndSet(false)) { + status = 500; + } else { + status = statusToReturn.get(); + } + 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(20)); + when(config.maxRetryBackoff()).thenReturn(Duration.ofSeconds(5)); + when(config.maxConcurrent()).thenReturn(4); + when(config.maxPending()).thenReturn(100); + when(config.requireHttps()).thenReturn(false); + } + + @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, meterRegistry); + listener.start(); + return listener; + } + + private static PolarisEvent testEvent() { + return new PolarisEvent( + PolarisEventType.AFTER_REFRESH_TABLE, + 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 deliversSignedJsonPayloadWithEnvelopeFields() throws Exception { + WebhookEventListener listener = createListener(); + try { + listener.onEvent(testEvent()); + + Awaitility.await("webhook delivered") + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); + + RecordedRequest request = recordedRequests.getFirst(); + JsonNode json = OBJECT_MAPPER.readTree(request.body()); + 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("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("realmid").asText()).isEqualTo(REALM); + assertThat(json.get("principal").asText()).isEqualTo(TEST_USER); + 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)); + assertThat( + meterRegistry + .find("polaris.event.webhook.deliveries") + .tag("result", "success") + .counter() + .count()) + .isEqualTo(1.0); + } 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(); + } finally { + listener.shutdown(); + } + } + + @Test + 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()); + + Awaitility.await("webhook delivered") + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(recordedRequests).hasSize(1)); + + 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(); + } + } + + @Test + void failsFastWhenEndpointNotConfigured() { + when(config.endpoint()).thenReturn(Optional.empty()); + WebhookEventListener listener = + new WebhookEventListener(config, OBJECT_MAPPER, CLOCK, meterRegistry); + assertThatThrownBy(listener::start) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("polaris.event-listener.webhook.endpoint"); + } + + @Test + 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 { + listener.onEvent(testEvent()); + + 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(); + } + } + + @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)); + } finally { + listener.shutdown(); + } + } + + @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, 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 + 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/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..97204a6321 100644 --- a/runtime/defaults/src/main/resources/application.properties +++ b/runtime/defaults/src/main/resources/application.properties @@ -173,6 +173,19 @@ 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 +# 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 # 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..ad79d64cc6 --- /dev/null +++ b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener_webhook.md @@ -0,0 +1,43 @@ +--- +# +# 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 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). + +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 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` | diff --git a/site/content/in-dev/unreleased/configuration/configuration-reference.md b/site/content/in-dev/unreleased/configuration/configuration-reference.md index b010b469cc..228105842b 100644 --- a/site/content/in-dev/unreleased/configuration/configuration-reference.md +++ b/site/content/in-dev/unreleased/configuration/configuration-reference.md @@ -138,6 +138,50 @@ 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 shaped after +the CloudEvents envelope (no CloudEvents SDK required): + +```json +{ + "specversion": "1.0", + "id": "11111111-2222-3333-4444-555555555555", + "type": "AFTER_REFRESH_TABLE", + "source": "org.apache.polaris", + "time": "2026-01-15T12:00:00Z", + "deliverytime": "2026-01-15T12:00:01Z", + "realmid": "my-realm", + "principal": "alice", + "activatedroles": ["role1", "role2"], + "requestid": "d3b0...", + "tableidentifier": "ns.tbl" +} +``` + +`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. 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. + +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 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 +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. + ### `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",