github-advanced-security[bot] commented on code in PR #20197:
URL: https://github.com/apache/druid/pull/20197#discussion_r4081760914
##########
processing/src/main/java/org/apache/druid/segment/loading/AcquireSegmentAction.java:
##########
@@ -19,76 +19,134 @@
package org.apache.druid.segment.loading;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.common.asyncresource.AsyncResource;
+import org.apache.druid.common.asyncresource.SettableAsyncResource;
import org.apache.druid.error.DruidException;
-import org.apache.druid.segment.ReferenceCountedObjectProvider;
-import org.apache.druid.segment.ReferenceCountedSegmentProvider;
+import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.segment.Segment;
import javax.annotation.Nullable;
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Supplier;
+import java.util.function.Consumer;
/**
- * This class represents an intent to acquire a reference to a {@link Segment}
and then use it to do stuff, and finally
- * closing when done.
+ * Handle for acquiring a reference to a {@link Segment} which might need to
be loaded on demand in the form of an
+ * {@link AsyncResource} delivering an {@link AcquireSegmentResult} whose
acquired segment reference carries every hold
+ * and reference associated with the acquisition inside its own {@link
Segment#close()}.
* <p>
- * When an {@link AcquireSegmentAction} is created, segment cache
implementations will create a
- * 'hold' to ensure it cannot be removed from the cache until this action has
been closed.
- * <p>
- * Calling {@link #getSegmentFuture()} is what actually initiates the
'action', and will return a segment reference
- * provider if already cached, or attempt to download from deep storage to
load into the cache if not. The
- * {@link Segment} returned by acquiring a reference from the {@link
ReferenceCountedObjectProvider} returned by the
- * future places a separate hold on the cache until the segment itself is
closed, and MUST be closed when the caller is
- * finished doing segment things with it.
+ * The load (if one is needed) starts when the handle is created; there is no
separate initiation step.
+ *
+ * <h3>Consumer protocol</h3>
+ * Register the handle with cleanup machinery (e.g. a {@code Closer})
immediately (this is safe at any lifecycle
+ * point). Wait for {@link #isReady()} via {@link #addReadyCallback} or {@link
#await}, then call {@link #release()}
+ * to take ownership of the {@link AcquireSegmentResult} (or surface the
producer's exception). After release, the
+ * caller owns closing the result (or the segment inside it); {@link #close()}
on this handle becomes a no-op.
* <p>
- * The caller must also call {@link #close()} on this object to clean up the
hold that exists while possibly loading
- * the segment, and may do so as soon as the {@link Segment} is acquired (or
can do so earlier to abort the load and
- * release the hold).
+ * Closing the handle without releasing cancels an in-flight load (releasing
any cache holds placed for it), or
+ * closes an already-delivered result. Closing before the result arrives fires
any pending ready callbacks so that
+ * waiting consumers learn the acquisition was aborted ({@link #get()} and
{@link #release()} then throw
+ * {@link
org.apache.druid.common.asyncresource.AsyncResourceCanceledException}). Note
that {@link #close()} is
+ * <b>not</b> idempotent (it throws if called twice), close exactly once.
*/
-public class AcquireSegmentAction implements Closeable
+public class AcquireSegmentAction extends
SettableAsyncResource<AcquireSegmentResult>
{
+ private static final Logger LOG = new Logger(AcquireSegmentAction.class);
+
+ /**
+ * Hook installed via {@link #setOnRelease}, run by {@link #release()} after
ownership transfers.
+ */
+ @Nullable
+ @GuardedBy("this")
+ private Consumer<AcquireSegmentResult> onRelease;
+
+ /**
+ * Handle representing a segment that is known to be missing: immediately
ready with
+ * {@link AcquireSegmentResult#empty()}.
+ */
public static AcquireSegmentAction missingSegment()
{
- return new AcquireSegmentAction(() ->
Futures.immediateFuture(AcquireSegmentResult.empty()), null);
+ return completed(AcquireSegmentResult.empty());
}
- private final Supplier<ListenableFuture<AcquireSegmentResult>>
segmentFutureSupplier;
- @Nullable
- private final Closeable loadCleanup;
- private final AtomicBoolean closed = new AtomicBoolean(false);
+ /**
+ * Handle that is immediately ready with the given result, for segments that
did not require an asynchronous load.
+ */
+ public static AcquireSegmentAction completed(AcquireSegmentResult result)
+ {
+ final AcquireSegmentAction action = new AcquireSegmentAction();
+ action.set(result);
+ return action;
+ }
- public AcquireSegmentAction(
- Supplier<ListenableFuture<AcquireSegmentResult>> segmentFutureSupplier,
- @Nullable Closeable loadCleanup
- )
+ /**
+ * Constructor for producers that install a canceler after creating whatever
the canceler must cancel (calling
+ * {@link #setCanceler} on a handle that has already become ready is a safe
no-op).
+ */
+ public AcquireSegmentAction()
+ {
+ super(true);
+ }
+
+ /**
+ * @param canceler optional callback invoked from {@link #close()} when the
handle is closed before the result has
+ * been delivered ({@link #set} or {@link #setException}).
Producers that support cancellation
+ * should provide one; producers that don't can pass {@code
null}, in which case {@link #close()}
+ * just stops observing the result.
+ */
+ public AcquireSegmentAction(@Nullable Runnable canceler)
{
- this.segmentFutureSupplier = segmentFutureSupplier;
- this.loadCleanup = loadCleanup;
+ this();
+ if (canceler != null) {
+ setCanceler(canceler);
+ }
}
/**
- * Get a {@link ReferenceCountedObjectProvider<Segment>} to acquire a
reference to an item that exists in the cache,
- * or if necessary/possible, fetching it from deep storage. This is
typically {@link ReferenceCountedSegmentProvider},
- * either as an immediate future if the segment already exists in cache. The
'action' to fetch the segment and return
- * the reference provider is not initiated until this method is called.
+ * Convenience setter: the result acts as its own closer, so closing an
un-released ready handle closes the
+ * delivered result. Returns false if this handle was closed before
delivery, in which case the producer retains
+ * ownership of the result and must close it.
*/
- public ListenableFuture<AcquireSegmentResult> getSegmentFuture()
+ public boolean set(AcquireSegmentResult result)
{
- if (closed.get()) {
- throw DruidException.defensive("Cannot getSegmentFuture() after
close()");
+ return super.set(result, result);
+ }
+
+ /**
+ * Installs a hook that {@link #release()} runs with the released result,
after ownership has transferred to the
+ * releasing caller. Since release can only succeed once, the hook runs at
most once; it never runs when the load
+ * fails or the handle is closed without release. Exceptions it throws are
logged and suppressed, so a broken hook
+ * cannot poison the delivery or leak the result.
+ * <p>
+ * Producers use this for exactly-once delivery accounting (e.g. counter
updates) without placing an obligation on
+ * consumers; install it before handing the action to the consumer.
+ */
+ public synchronized void setOnRelease(Consumer<AcquireSegmentResult> hook)
+ {
+ if (onRelease != null) {
+ throw DruidException.defensive("onRelease hook already set, cannot call
setOnRelease()");
}
- return segmentFutureSupplier.get();
+ onRelease = hook;
}
- @Override
- public void close() throws IOException
+ @Override // Overridden to change access from protected to public
+ public AcquireSegmentResult release()
Review Comment:
## CodeQL / Non-synchronized override of synchronized method
Method 'release' overrides a synchronized method in
[org.apache.druid.common.asyncresource.SettableAsyncResource](1) but is not
synchronized.
[Show more
details](https://github.com/apache/druid/security/code-scanning/12002)
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
