Skip to content

fix: handle null states in EVCacheBulkGetFuture and refactor #179

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shy-1234 merged 5 commits into from
Feb 5, 2026
Merged

fix: handle null states in EVCacheBulkGetFuture and refactor #179

Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a1de62b
fix: handle null states in EVCacheBulkGetFuture to prevent NPE
Aug 15, 2025
00ff306
fix EVCacheBulkGetFuture
Jan 23, 2026
e93ef48
fix opsArray should be initialized later
Jan 23, 2026
7eb75e8
let's make less refactoring at this point..
Feb 3, 2026
f14eaa1
fix handleBulkException behavior
Feb 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 60 additions & 18 deletions evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheBulkGetFuture.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,8 @@ public Map<String, T> getSome(long to, TimeUnit unit, boolean throwException, bo
throws InterruptedException, ExecutionException {
assert operationStates != null;

// Note: The latch here is counterintuitive. Based on the implementation in EVCacheMemcachedClient.asyncGetBulk(),
// the latch count is set to 1 no matter the chunk size (when > 0) and only decrement when pendingChunks counts down to 0.
boolean allCompleted = latch.await(to, unit);
if(log.isDebugEnabled()) log.debug("Took " + (System.currentTimeMillis() - start)+ " to fetch " + rvMap.size() + " keys from " + client);
long pauseDuration = -1;
Expand Down Expand Up @@ -120,25 +122,44 @@ public Map<String, T> getSome(long to, TimeUnit unit, boolean throwException, bo
}

boolean hadTimedoutOp = false;
Operation[] opsArray = ops.toArray(new Operation[0]);
for (int i = 0; i < operationStates.length(); i++) {
SingleOperationState state = operationStates.get(i);
if (!state.completed && !allCompleted) {
MemcachedConnection.opTimedOut(state.op);

if (state == null) {
// Operation not yet signaled completion (cancel should still trigger completion) ==> latch timed out
// This also indicates allCompleted == false because the latch count wouldn't have drop to 0.
Operation op = opsArray[i];
op.timeOut();
MemcachedConnection.opTimedOut(op);
hadTimedoutOp = true;
} else {
MemcachedConnection.opSucceeded(state.op);
if (state.timedOut) {
MemcachedConnection.opTimedOut(state.op);
hadTimedoutOp = true;
} else {
MemcachedConnection.opSucceeded(state.op);
}
}
}

if (!allCompleted && !hasZF && hadTimedoutOp) statusString = EVCacheMetricsFactory.TIMEOUT;
if (hadTimedoutOp && !hasZF) statusString = EVCacheMetricsFactory.TIMEOUT;

@Sunjeet Sunjeet Jan 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it intentional to omit && hadTimedoutOp in the conditional?

@shy-1234 shy-1234 Jan 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I assume you mean omit !allCompleted? This is based on the fact that "!allCompleted" will always lead to hadTimedoutOp.

// Should we throw when timeout?

for (int i = 0; i < operationStates.length(); i++) {
SingleOperationState state = operationStates.get(i);
if (state.cancelled) {
if (hasZF) statusString = EVCacheMetricsFactory.CANCELLED;
if (throwException) throw new ExecutionException(new CancellationException("Cancelled"));

// state == null always means timed out and was handled.
if (state != null) {
if (state.cancelled) {
if (hasZF) statusString = EVCacheMetricsFactory.CANCELLED;
if (throwException) throw new ExecutionException(new CancellationException("Cancelled"));
}
if (state.errored) {
if (hasZF) statusString = EVCacheMetricsFactory.ERROR;
if (throwException) throw new ExecutionException(state.op.getException());
}
}
if (state.errored && throwException) throw new ExecutionException(state.op.getException());
}

Map<String, T> m = new HashMap<String, T>();
Expand Down Expand Up @@ -219,20 +240,27 @@ public CompletableFuture<Map<String, T>> getAsyncSome(long timeout, TimeUnit uni

public void handleBulkException() {
ExecutionException t = null;
Operation[] opsArray = ops.toArray(new Operation[0]);
for (int i = 0; i < operationStates.length(); i++) {
SingleOperationState state = operationStates.get(i);
if (!state.completed) {

if (state == null) {
Operation op = opsArray[i];
op.timeOut();

@Sunjeet Sunjeet Jan 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we want to signal timeout here? Iiuc we're handling that one of the ops in the bulk had an exception, shouldn't we signal a cancellation instead?

@shy-1234 shy-1234 Jan 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree. signal a cancellation to the op seems better if we enter here while state is still null

@shy-1234 shy-1234 Jan 23, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hmm after second look at this method, I wonder what's the intention of this method. The caller caught an exception, but we are not doing anything with that "ex". Instead we throw based on whichever ops we found "first" that is either Cancelled or TimedOut or Errored...
I feel like the correct implementation should be something like

    public void handleBulkException(Throwable ex) {
        Operation[] opsArray = ops.toArray(new Operation[0]);
        for (int i = 0; i < operationStates.length(); i++) {
            SingleOperationState state = operationStates.get(i);

            if (state == null) {
                Operation op = opsArray[i];
                op.cancel();
                MemcachedConnection.opSucceeded(op);
            } else {
                // Use pre-collected state
                if (state.timedOut) {
                    MemcachedConnection.opTimedOut(state.op);
                } else {
                    MemcachedConnection.opSucceeded(state.op);
                }
            }
        }

        throw new RuntimeException(ex);
    }

MemcachedConnection.opTimedOut(op);
t = new ExecutionException(new CheckedOperationTimeoutException("Checked Operation timed out.", op));
} else {
// Use pre-collected state
if (state.cancelled) {
throw new RuntimeException(new ExecutionException(new CancellationException("Cancelled")));
} else if (state.errored) {
throw new RuntimeException(new ExecutionException(state.op.getException()));
} else {
state.op.timeOut();
} else if (state.timedOut) {
MemcachedConnection.opTimedOut(state.op);
t = new ExecutionException(new CheckedOperationTimeoutException("Checked Operation timed out.", state.op));
} else {
MemcachedConnection.opSucceeded(state.op);
}
} else {
MemcachedConnection.opSucceeded(state.op);
}
}

Expand All @@ -257,19 +285,33 @@ public void doAsyncGetSome(CompletableFuture<Map<String, T>> promise) {
public Single<Map<String, T>> getSome(long to, TimeUnit units, boolean throwException, boolean hasZF, Scheduler scheduler) {
return observe().timeout(to, units, Single.create(subscriber -> {
try {
Operation[] opsArray = ops.toArray(new Operation[0]);
for (int i = 0; i < operationStates.length(); i++) {
SingleOperationState state = operationStates.get(i);
if (!state.completed) {
MemcachedConnection.opTimedOut(state.op);

if (state == null) {
Operation op = opsArray[i];
op.timeOut();
MemcachedConnection.opTimedOut(op);
// Should we throw when timeout?
} else {
MemcachedConnection.opSucceeded(state.op);
if (state.timedOut) {
MemcachedConnection.opTimedOut(state.op);
// Should we throw when timeout?
} else {
MemcachedConnection.opSucceeded(state.op);
}
}
}

for (int i = 0; i < operationStates.length(); i++) {
SingleOperationState state = operationStates.get(i);
if (state.cancelled && throwException) throw new ExecutionException(new CancellationException("Cancelled"));
if (state.errored && throwException) throw new ExecutionException(state.op.getException());

// state == null always means timed out and was handled.
if (state != null) {
if (state.cancelled && throwException) throw new ExecutionException(new CancellationException("Cancelled"));
if (state.errored && throwException) throw new ExecutionException(state.op.getException());
}
}

Map<String, T> m = new HashMap<String, T>();
Expand Down