dimas-b commented on code in PR #4061: URL: https://github.com/apache/polaris/pull/4061#discussion_r3012579135
########## runtime/service/src/main/java/org/apache/polaris/service/context/catalog/RequestIdHolder.java: ########## @@ -0,0 +1,66 @@ +/* + * 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.service.context.catalog; + +import io.quarkus.arc.Unremovable; +import jakarta.annotation.Nullable; +import jakarta.enterprise.context.RequestScoped; +import jakarta.enterprise.inject.Produces; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.polaris.core.context.RequestIdSupplier; + +/** + * Request-scoped holder for the request ID. + * + * <p>On HTTP request threads the request ID is set by {@code RequestIdFilter}. On async task + * threads a new CDI request scope is activated by {@code TaskExecutorImpl}, and the request ID from + * the originating request is propagated into this holder by {@code RequestIdPropagator}. + * + * <p>The holder exposes the stored ID as a {@link RequestIdSupplier} produced bean so that any + * component injecting {@code Instance<RequestIdSupplier>} can resolve it without depending on + * JAX-RS internals. + */ +@RequestScoped +public class RequestIdHolder { + + private final AtomicReference<String> requestId = new AtomicReference<>(); + + @Nullable + public String get() { + return requestId.get(); + } + + /** + * Produces a {@link RequestIdSupplier} for the current request scope. The returned supplier reads + * the request ID from this holder at the time it is called, so it reflects any value set after + * this method is first invoked. + */ + @Produces + @RequestScoped + @Unremovable Review Comment: Why `@Unremovable`? ########## runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventMetadataFactory.java: ########## @@ -23,24 +23,23 @@ import io.quarkus.security.identity.CurrentIdentityAssociation; import io.quarkus.security.identity.SecurityIdentity; import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.context.ContextNotActiveException; import jakarta.enterprise.inject.Instance; import jakarta.inject.Inject; import java.time.Clock; import java.util.Map; import java.util.Optional; import org.apache.polaris.core.auth.PolarisPrincipal; import org.apache.polaris.core.context.RealmContext; -import org.apache.polaris.service.tracing.RequestIdFilter; -import org.jboss.resteasy.reactive.server.core.CurrentRequestManager; -import org.jboss.resteasy.reactive.server.core.ResteasyReactiveRequestContext; -import org.jboss.resteasy.reactive.server.jaxrs.ContainerRequestContextImpl; +import org.apache.polaris.core.context.RequestIdSupplier; @ApplicationScoped public class PolarisEventMetadataFactory { @Inject Clock clock; @Inject CurrentIdentityAssociation currentIdentityAssociation; @Inject Instance<RealmContext> realmContext; + @Inject RequestIdSupplier requestIdSupplier; Review Comment: Why not `RequestIdHolder`?.. It's simpler, no? ########## runtime/service/src/main/java/org/apache/polaris/service/task/AsyncContextPropagator.java: ########## @@ -0,0 +1,69 @@ +/* + * 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.service.task; + +import jakarta.annotation.Nonnull; +import jakarta.annotation.Nullable; + +/** + * Extension point for propagating request-scoped context across the async task boundary. + * + * <p>Each implementation is responsible for a single piece of request-scoped context (e.g. realm + * identity, authenticated principal, request ID). Implementations are CDI beans (typically + * {@code @ApplicationScoped}). {@link TaskExecutorImpl} discovers all implementations via CDI + * {@code Instance} injection, so adding a new propagation concern requires only a new bean — no + * existing code needs to change. + * + * <p>Lifecycle: + * + * <ol> + * <li>{@link #capture()} is called on the request thread (active request scope). The + * implementation reads its relevant context and returns an opaque snapshot. + * <li>The snapshot is carried across the async boundary together with the propagator that created + * it. + * <li>{@link #restore(Object)} is called inside the task thread's new CDI request scope. The + * implementation re-establishes its context from the snapshot and returns an {@link + * AutoCloseable} for cleanup after the task finishes (e.g. MDC restoration). + * </ol> + */ +public interface AsyncContextPropagator { + + /** + * Captures relevant context from the current request scope. + * + * <p>The returned snapshot may be restored multiple times across retries and different threads. + * Implementations must ensure the captured state is <strong>immutable</strong> and + * <strong>thread-safe</strong>. + * + * @return an opaque snapshot that will be passed to {@link #restore(Object)} in the task thread, + * or {@code null} if no context is available to capture. + */ + @Nullable + Object capture(); + + /** + * Restores the captured context into the task thread's active request scope. + * + * @param capturedState the snapshot returned by {@link #capture()}, may be {@code null}. + * @return an {@link AutoCloseable} that is closed after the task finishes. Implementations that + * need no cleanup must return a no-op ({@code () -> {}}). Must not return {@code null}. + */ + @Nonnull + AutoCloseable restore(@Nullable Object capturedState); Review Comment: I'm not sure why context restoration should require a closable follow-up action. It seems to complicate the design and intended use case. The only currently relevant use case deals with MDC, but I believe it would be preferable to keep MDC out of the (CDI) context propagation code. It's fine to populate MDC where necessary by specific code (e.g. in `TaskExecutorImpl`). WDYT? ########## runtime/service/src/main/java/org/apache/polaris/service/task/RealmContextPropagator.java: ########## @@ -0,0 +1,82 @@ +/* + * 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.service.task; + +import jakarta.annotation.Nullable; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.context.ContextNotActiveException; +import jakarta.inject.Inject; +import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.service.context.catalog.RealmContextHolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Propagates the realm context across the async task boundary via {@link RealmContextHolder}. + * + * <p>The full {@link RealmContext} object is captured — not just the realm identifier string — so + * that vendor-specific {@code RealmContext} implementations (which may carry additional routing + * information) survive the async boundary intact. + * + * <p>At capture time {@link RealmContextHolder} is request-scoped; its CDI proxy resolves to the + * holder in the currently active request scope. On a normal HTTP request thread that scope holds the + * realm set by the request filter. When an async task handler schedules a follow-up task (no active + * JAX-RS request), {@code RealmContextHolder} in that task's scope already contains the realm + * restored by this propagator's {@link #restore} path, so capture continues to work correctly for + * nested task submission. + */ +@ApplicationScoped +public class RealmContextPropagator implements AsyncContextPropagator { + + private static final Logger LOGGER = LoggerFactory.getLogger(RealmContextPropagator.class); + + private final RealmContextHolder realmContextHolder; + + @SuppressWarnings("unused") // Required by CDI + protected RealmContextPropagator() { + this(null); + } + + @Inject + public RealmContextPropagator(RealmContextHolder realmContextHolder) { + this.realmContextHolder = realmContextHolder; + } + + @Nullable + @Override + public Object capture() { Review Comment: This has implications to the `AutoClosable` returned from `restore()` (separate comment thread). If you agree to the general principles, I believe it would be preferable to do the "state" refactoring in this PR. I hope it can resolve my other concerns "automatically" too. ########## runtime/service/src/main/java/org/apache/polaris/service/task/PrincipalContextPropagator.java: ########## @@ -0,0 +1,83 @@ +/* + * 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.service.task; + +import jakarta.annotation.Nullable; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.context.ContextNotActiveException; +import jakarta.enterprise.inject.Instance; +import jakarta.inject.Inject; +import org.apache.polaris.core.auth.ImmutablePolarisPrincipal; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.service.context.catalog.PolarisPrincipalHolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Propagates the authenticated principal across the async task boundary via {@link + * PolarisPrincipalHolder}. + * + * <p>A clone of the principal is captured at submission time so the task thread uses a stable + * snapshot that is independent of the originating request scope's lifecycle. + */ +@ApplicationScoped +public class PrincipalContextPropagator implements AsyncContextPropagator { + + private static final Logger LOGGER = LoggerFactory.getLogger(PrincipalContextPropagator.class); + + private final PolarisPrincipalHolder polarisPrincipalHolder; + private final Instance<PolarisPrincipal> polarisPrincipal; + + @SuppressWarnings("unused") // Required by CDI + protected PrincipalContextPropagator() { + this(null, null); + } + + @Inject + public PrincipalContextPropagator( + PolarisPrincipalHolder polarisPrincipalHolder, Instance<PolarisPrincipal> polarisPrincipal) { + this.polarisPrincipalHolder = polarisPrincipalHolder; + this.polarisPrincipal = polarisPrincipal; + } + + @Nullable + @Override + public Object capture() { + PolarisPrincipal clone = null; + if (polarisPrincipal.isResolvable()) { + try { + // Clone to allow task thread get a stable snapshot regardless of the request scope lifecycle. + clone = ImmutablePolarisPrincipal.builder().from(polarisPrincipal.get()).build(); + } catch (ContextNotActiveException e) { + // scope not active, return null + } + } + LOGGER.trace("capture principal={}", clone != null ? clone.getName() : null); + return clone; + } + + @Override + public AutoCloseable restore(@Nullable Object capturedState) { + LOGGER.trace("restore principal={}", capturedState != null ? ((PolarisPrincipal) capturedState).getName() : null); + if (capturedState != null) { + polarisPrincipalHolder.set((PolarisPrincipal) capturedState); Review Comment: So it's another reason to use the state pattern in this PR (as opposed to follow-up), I think 🤔 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
