-
Notifications
You must be signed in to change notification settings - Fork 27
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.
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
Fix numWorkItemsArePending bug #1102
Changes from 3 commits
31ae47d
e48005b
ab9b653
1a926d1
f573157
7dc4f28
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -397,14 +397,11 @@ | |
} | ||
} | ||
|
||
private int numWorkItemsArePending( | ||
int maxItemsToCheckFor, | ||
private int numWorkItemsArePendingInternal( | ||
Supplier<IWorkCoordinationContexts.IPendingWorkItemsContext> contextSupplier | ||
) throws IOException, InterruptedException { | ||
try (var context = contextSupplier.get()) { | ||
refresh(context::getRefreshContext); | ||
// TODO: Switch this to use _count | ||
log.warn("Switch this to use _count"); | ||
final var queryBody = "{\n" | ||
+ "\"query\": {" | ||
+ " \"bool\": {" | ||
|
@@ -419,36 +416,35 @@ | |
+ " }" | ||
+ " ]" | ||
+ " }" | ||
+ "}" | ||
+ "}," | ||
+ "\"size\": 1" | ||
+ "}"; | ||
|
||
var path = INDEX_NAME + "/_search" + (maxItemsToCheckFor <= 0 ? "" : "?size=" + maxItemsToCheckFor); | ||
var path = INDEX_NAME + "/_search"; | ||
var response = httpClient.makeJsonRequest(AbstractedHttpClient.POST_METHOD, path, null, queryBody); | ||
|
||
final var resultHitsUpper = objectMapper.readTree(response.getPayloadBytes()).path("hits"); | ||
var statusCode = response.getStatusCode(); | ||
if (statusCode != 200) { | ||
throw new IllegalStateException( | ||
"Querying for pending (expired or not) work, " | ||
+ "returned an unexpected status code " | ||
+ statusCode | ||
+ " instead of 200" | ||
"Querying for pending (expired or not) work, " | ||
+ "returned an unexpected status code " | ||
+ statusCode | ||
+ " instead of 200" | ||
); | ||
} | ||
return resultHitsUpper.path("hits").size(); | ||
return objectMapper.readTree(response.getPayloadBytes()).path("hits").path("total").path("value").asInt(); | ||
} | ||
} | ||
|
||
@Override | ||
public int numWorkItemsArePending(Supplier<IWorkCoordinationContexts.IPendingWorkItemsContext> contextSupplier) | ||
throws IOException, InterruptedException { | ||
return numWorkItemsArePending(-1, contextSupplier); | ||
return numWorkItemsArePendingInternal(contextSupplier); | ||
Check warning on line 441 in RFS/src/main/java/org/opensearch/migrations/bulkload/workcoordination/OpenSearchWorkCoordinator.java Codecov / codecov/patchRFS/src/main/java/org/opensearch/migrations/bulkload/workcoordination/OpenSearchWorkCoordinator.java#L441
|
||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. how do you get back the right value for this if you're running the query w/ size=1? |
||
} | ||
|
||
@Override | ||
public boolean workItemsArePending(Supplier<IWorkCoordinationContexts.IPendingWorkItemsContext> contextSupplier) | ||
throws IOException, InterruptedException { | ||
return numWorkItemsArePending(1, contextSupplier) >= 1; | ||
return numWorkItemsArePendingInternal(contextSupplier) >= 1; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is a very different call now. My original intention was that in this case, I really only needed to know if the set was empty or not. Counting up thousands of documents won't be necessary. If -1 wasn't working to return the whole list, we should probably just have two separate functions - isEmpty() and count(). There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Okay, I ran some tests with a couple thousand items in the index (And it turns out that adding 3790 items pretty quickly basically completely locks up a testcontainer cluster -- even with 30 second sleeps between attempts, I haven't been able to add a single document to it in >10 minutes -- our worst case situations for blocking indices are really bad). The tests here are 1/
Conveniently
|
||
} | ||
|
||
enum UpdateResult { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For future reference: The author of this P found that _count was slower than the search