Skip to content

fix: Prevent NPE in query_lookup when hit index differs from configur… #7098

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

Open
nagendramohan wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix: Prevent NPE in query_lookup when hit index differs from configur… #7098

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a278c27
fix: Prevent NPE in query_lookup when hit index differs from configur…
nagendramohan Aug 15, 2026
632ea79
refactor: Build ordered index keys before msearch builder lambda
nagendramohan Aug 17, 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
141 changes: 88 additions & 53 deletions ...rg/opensearch/dataprepper/plugins/sink/opensearch/index/ExistingDocumentQueryManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import org.opensearch.client.opensearch._types.query_dsl.TermsQueryField;
import org.opensearch.client.opensearch.core.MsearchRequest;
import org.opensearch.client.opensearch.core.MsearchResponse;
import org.opensearch.client.opensearch.core.msearch.MultiSearchResponseItem;
import org.opensearch.dataprepper.metrics.PluginMetrics;
import org.opensearch.dataprepper.plugins.sink.opensearch.BulkOperationWrapper;
import org.opensearch.dataprepper.plugins.sink.opensearch.index.model.QueryManagerBulkOperation;
Expand All @@ -19,6 +20,7 @@

import java.time.Duration;
import java.time.Instant;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
Expand Down Expand Up @@ -127,12 +129,16 @@ public void run() {
void runQueryLoop() {
if (!bulkOperationsWaitingForQuery.isEmpty() && documentsCurrentlyBeingQueriedGauge.get() > 0) {

// Query for existing documents
final MsearchRequest msearchRequest = buildMultiSearchRequest();
// Query for existing documents. Track the index key used for each search request, in request
// order, so responses can be correlated back to the correct pending-operations map. We cannot
// rely on hit.index() for this: a search against an alias (or datastream) returns the concrete
// backing index in the hit, which differs from the configured index/alias key. See #6902.
final List<String> searchRequestIndexKeys = new ArrayList<>();
final MsearchRequest msearchRequest = buildMultiSearchRequest(searchRequestIndexKeys);
final MsearchResponse<ObjectNode> msearchResponse = queryForTermValues(msearchRequest);

// Drop and Release Existing Documents
dropAndReleaseFoundEvents(msearchResponse);
dropAndReleaseFoundEvents(msearchResponse, searchRequestIndexKeys);

// Move non-existing documents past query_duration to bulkOperationsReadyForIndex
moveBulkRequestsThatHaveReachedQueryDuration();
Expand Down Expand Up @@ -183,30 +189,43 @@ public Set<BulkOperationWrapper> getAndClearBulkOperationsReadyToIndex() {
}
}

private MsearchRequest buildMultiSearchRequest() {
private MsearchRequest buildMultiSearchRequest(final List<String> orderedIndexKeys) {
// Build the (index key, term-value chunk) pairs deterministically BEFORE constructing the request,
// so the order of orderedIndexKeys is guaranteed to align with the order in which searches are
// added to the msearch request (and therefore with the response order), independent of how or when
// the builder lambda below is invoked. This alignment is what lets dropAndReleaseFoundEvents map a
// response back to its configured index/alias key rather than the concrete backing index reported
// by hit.index(). See #6902.
final List<Map.Entry<String, List<FieldValue>>> searchRequestChunks = new ArrayList<>();
for (final Map.Entry<String, Map<String, QueryManagerBulkOperation>> entry : bulkOperationsWaitingForQuery.entrySet()) {
final String index = entry.getKey();
final List<FieldValue> values = getTermValues(entry.getValue().values());
final int batchSize = 1000;

LOG.info("Creating search requests for {} query term values in batches of {}", values.size(), batchSize);
for (int i = 0; i < values.size(); i += batchSize) {
final List<FieldValue> chunk = values.subList(i, Math.min(i + batchSize, values.size()));
orderedIndexKeys.add(index);
searchRequestChunks.add(new AbstractMap.SimpleEntry<>(index, chunk));
}
}

return MsearchRequest.of(m -> {
for (final Map.Entry<String, Map<String, QueryManagerBulkOperation>> entry : bulkOperationsWaitingForQuery.entrySet()) {
final String index = entry.getKey();
final List<FieldValue> values = getTermValues(entry.getValue().values());
final int batchSize = 1000;

LOG.info("Creating search requests for {} query term values in batches of {}", values.size(), batchSize);
for (int i = 0; i < values.size(); i += batchSize) {
final List<FieldValue> chunk = values.subList(i, Math.min(i + batchSize, values.size()));

m.searches(s -> s
.header(h -> h.index(index))
.body(b -> b
.size(chunk.size() * 2)
.source(source -> source.filter(f -> f.includes(queryTerm)))
.query(Query.of(q -> q
.terms(TermsQuery.of(t -> t
.field(queryTerm)
.terms(TermsQueryField.of(tf -> tf.value(chunk)))
))
))
));
}
for (final Map.Entry<String, List<FieldValue>> searchRequestChunk : searchRequestChunks) {
final String index = searchRequestChunk.getKey();
final List<FieldValue> chunk = searchRequestChunk.getValue();
m.searches(s -> s
.header(h -> h.index(index))
.body(b -> b
.size(chunk.size() * 2)
.source(source -> source.filter(f -> f.includes(queryTerm)))
.query(Query.of(q -> q
.terms(TermsQuery.of(t -> t
.field(queryTerm)
.terms(TermsQueryField.of(tf -> tf.value(chunk)))
))
))
));
}
return m;
});
Expand Down Expand Up @@ -257,36 +276,52 @@ private void moveBulkRequestsThatHaveReachedQueryDuration() {
}
}

private void dropAndReleaseFoundEvents(final MsearchResponse<ObjectNode> msearchResponse) {
msearchResponse.responses().forEach(response -> {
private void dropAndReleaseFoundEvents(final MsearchResponse<ObjectNode> msearchResponse, final List<String> orderedIndexKeys) {
final List<MultiSearchResponseItem<ObjectNode>> responses = msearchResponse.responses();
for (int responseIndex = 0; responseIndex < responses.size(); responseIndex++) {
final MultiSearchResponseItem<ObjectNode> response = responses.get(responseIndex);
if (response.isFailure()) {
LOG.error("Search response failed, potential for duplicate documents: {}", response.failure().error().toString());
} else {
response.result().hits().hits().forEach(hit -> {
final String indexForHit = hit.index();
final ObjectNode sourceForHit = hit.source();
final String queryTermValue = sourceForHit.findValue(queryTerm).textValue();
continue;
}

lockWaitingForQuery.lock();
try {
final Map<String, QueryManagerBulkOperation> bulkOperationsForIndex = bulkOperationsWaitingForQuery.get(indexForHit);
final QueryManagerBulkOperation bulkOperationToRelease = bulkOperationsForIndex.get(queryTermValue);
if (bulkOperationToRelease == null) {
// Means two documents with the same query term value were found
LOG.warn("Bulk operation for term value {} with id {} is null, potentially a duplicate document", queryTermValue, hit.id());
potentialDuplicatesDeleted.increment();
} else {
LOG.debug("Found document with query term {}, dropping and releasing Event handle", queryTermValue);
bulkOperationToRelease.getBulkOperationWrapper().releaseEventHandle(true);
eventsDroppedAndReleasedCounter.increment();
documentsCurrentlyBeingQueriedGauge.decrementAndGet();
bulkOperationsForIndex.remove(queryTermValue);
}
} finally {
lockWaitingForQuery.unlock();
// Correlate this response to the index key of the request that produced it (msearch preserves
// request order). hit.index() must not be used here: for aliases/datastreams the hit reports the
// concrete backing index, which differs from the configured index/alias key and would never match
// the pending-operations map, causing an NPE that wedges the query loop indefinitely. See #6902.
final String indexKey = responseIndex < orderedIndexKeys.size() ? orderedIndexKeys.get(responseIndex) : null;
final Map<String, QueryManagerBulkOperation> bulkOperationsForIndex =
indexKey == null ? null : bulkOperationsWaitingForQuery.get(indexKey);

response.result().hits().hits().forEach(hit -> {
final ObjectNode sourceForHit = hit.source();
final String queryTermValue = sourceForHit.findValue(queryTerm).textValue();

lockWaitingForQuery.lock();
try {
if (bulkOperationsForIndex == null) {
// Defensive: no pending operations tracked for this index key. Skip rather than
// throwing so a single unexpected response can never NPE and block the query loop.
LOG.warn("No pending bulk operations found for index key {} (hit index {}); skipping release for term value {}",
indexKey, hit.index(), queryTermValue);
return;
}
});
}
});
final QueryManagerBulkOperation bulkOperationToRelease = bulkOperationsForIndex.get(queryTermValue);
if (bulkOperationToRelease == null) {
// Means two documents with the same query term value were found
LOG.warn("Bulk operation for term value {} with id {} is null, potentially a duplicate document", queryTermValue, hit.id());
potentialDuplicatesDeleted.increment();
} else {
LOG.debug("Found document with query term {}, dropping and releasing Event handle", queryTermValue);
bulkOperationToRelease.getBulkOperationWrapper().releaseEventHandle(true);
eventsDroppedAndReleasedCounter.increment();
documentsCurrentlyBeingQueriedGauge.decrementAndGet();
bulkOperationsForIndex.remove(queryTermValue);
}
} finally {
lockWaitingForQuery.unlock();
}
});
}
}
}
64 changes: 60 additions & 4 deletions ...pensearch/dataprepper/plugins/sink/opensearch/index/ExistingDocumentQueryManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
Expand Down Expand Up @@ -107,7 +108,7 @@ void add_bulk_operation_and_found_in_query_drops_and_releases_event() throws IOE
final MultiSearchItem<ObjectNode> multiSearchItem = mock(MultiSearchItem.class);
final HitsMetadata<ObjectNode> hitsMetadata = mock(HitsMetadata.class);
final Hit<ObjectNode> hit = mock(Hit.class);
when(hit.index()).thenReturn(index);
lenient().when(hit.index()).thenReturn(index);

final ObjectNode objectNode = mock(ObjectNode.class);
final JsonNode jsonNode = mock(JsonNode.class);
Expand Down Expand Up @@ -141,6 +142,61 @@ void add_bulk_operation_and_found_in_query_drops_and_releases_event() throws IOE
verifyNoMoreInteractions(indexConfiguration);
}

@Test
void add_bulk_operation_and_found_in_query_when_hit_index_differs_from_configured_index_drops_and_releases_event() throws IOException {
// Regression test for #6902: when the sink writes to an alias (or datastream), the search hit reports
// the concrete backing index (e.g. "my-alias-000001"), which differs from the configured index/alias
// key ("my-alias"). Previously dropAndReleaseFoundEvents looked up the pending operations by
// hit.index(), which returned null and threw a NullPointerException, wedging the query loop and
// leaving the affected documents permanently stuck (never indexed, dropped, or sent to the DLQ).
final BulkOperationWrapper bulkOperationWrapper = mock(BulkOperationWrapper.class);
final String configuredIndex = UUID.randomUUID().toString();
final String concreteBackingIndex = configuredIndex + "-000001";
final String termValue = UUID.randomUUID().toString();
when(bulkOperationWrapper.getTermValue()).thenReturn(termValue);
when(bulkOperationWrapper.getIndex()).thenReturn(configuredIndex);

final MsearchResponse<ObjectNode> msearchResponse = mock(MsearchResponse.class);
final MultiSearchResponseItem<ObjectNode> responseItem = mock(MultiSearchResponseItem.class);
when(responseItem.isFailure()).thenReturn(false);

final MultiSearchItem<ObjectNode> multiSearchItem = mock(MultiSearchItem.class);
final HitsMetadata<ObjectNode> hitsMetadata = mock(HitsMetadata.class);
final Hit<ObjectNode> hit = mock(Hit.class);
// The hit carries the concrete backing index, not the configured alias key.
lenient().when(hit.index()).thenReturn(concreteBackingIndex);

final ObjectNode objectNode = mock(ObjectNode.class);
final JsonNode jsonNode = mock(JsonNode.class);
when(jsonNode.textValue()).thenReturn(termValue);
when(objectNode.findValue(queryTerm)).thenReturn(jsonNode);
when(hit.source()).thenReturn(objectNode);

when(multiSearchItem.hits()).thenReturn(hitsMetadata);
when(hitsMetadata.hits()).thenReturn(List.of(hit));

when(responseItem.result()).thenReturn(multiSearchItem);

when(msearchResponse.responses()).thenReturn(List.of(responseItem));

when(openSearchClient.msearch(any(MsearchRequest.class), eq(ObjectNode.class)))
.thenReturn(msearchResponse);

final ExistingDocumentQueryManager objectUnderTest = createObjectUnderTest();

objectUnderTest.addBulkOperation(bulkOperationWrapper);
when(documentsCurrentlyQueried.get()).thenReturn(1);

// Must not throw NullPointerException, and must drop + release the found duplicate.
objectUnderTest.runQueryLoop();

verify(eventsDroppedAndReleased).increment();
verify(eventsAddedForQuerying).increment();
verify(documentsCurrentlyQueried).incrementAndGet();
verify(documentsCurrentlyQueried).decrementAndGet();
verify(bulkOperationWrapper).releaseEventHandle(true);
}

@Test
void add_bulk_operation_and_not_found_in_query_returns_as_ready_to_ingest() throws IOException, InterruptedException, NoSuchFieldException, IllegalAccessException {
when(indexConfiguration.getQueryDuration()).thenReturn(Duration.ofMillis(1));
Expand All @@ -158,7 +214,7 @@ void add_bulk_operation_and_not_found_in_query_returns_as_ready_to_ingest() thro
final MultiSearchItem<ObjectNode> multiSearchItem = mock(MultiSearchItem.class);
final HitsMetadata<ObjectNode> hitsMetadata = mock(HitsMetadata.class);
final Hit<ObjectNode> hit = mock(Hit.class);
when(hit.index()).thenReturn(index);
lenient().when(hit.index()).thenReturn(index);

final ObjectNode objectNode = mock(ObjectNode.class);
final JsonNode jsonNode = mock(JsonNode.class);
Expand Down Expand Up @@ -261,7 +317,7 @@ void query_response_with_two_documents_with_same_term_value_tracks_duplicate_doc
final MultiSearchItem<ObjectNode> multiSearchItem = mock(MultiSearchItem.class);
final HitsMetadata<ObjectNode> hitsMetadata = mock(HitsMetadata.class);
final Hit<ObjectNode> hit = mock(Hit.class);
when(hit.index()).thenReturn(index);
lenient().when(hit.index()).thenReturn(index);

final ObjectNode objectNode = mock(ObjectNode.class);
final JsonNode jsonNode = mock(JsonNode.class);
Expand All @@ -270,7 +326,7 @@ void query_response_with_two_documents_with_same_term_value_tracks_duplicate_doc
when(hit.source()).thenReturn(objectNode);

final Hit<ObjectNode> duplicateHit = mock(Hit.class);
when(duplicateHit.index()).thenReturn(index);
lenient().when(duplicateHit.index()).thenReturn(index);
when(duplicateHit.id()).thenReturn(UUID.randomUUID().toString());

when(jsonNode.textValue()).thenReturn(termValue);
Expand Down
Loading