[BUG] NullPointerException in ExistingDocumentQueryManager.dropAndReleaseFoundEvents when query_lookup finds existing documents — documents permanently stuck, pipeline effectively blocked

Describe the bug
When using query_lookup in the OpenSearch sink and a duplicate document is found (i.e. a document with the same query_term value already exists in the index), ExistingDocumentQueryManager.dropAndReleaseFoundEvents throws a NullPointerException because bulkOperationsForIndex is null at line 273.
Despite the log message saying "retrying", the pipeline does not recover. The same documents remain permanently stuck in the query buffer and trigger the same NPE on every subsequent cycle indefinitely. The affected documents are never indexed, never dropped, and never sent to the DLQ. The only recovery is a full pipeline restart, which loses the buffered events.

To Reproduce

1. Configure an OpenSearch sink with query_lookup using a pre-computed event field as query_term:

processor:
  - add_entries:
      entries:
        - key: "dedup_key"
          value_expression: '/source + "_" + /articleId'
sink:
  - opensearch:
      hosts: ["http://opensearch-host:9200"]
      index: my-alias
      index_type: custom
      document_id: ${/dedup_key}
      query_lookup:
        query_when: '/articleId != null and /source != null'
        query_term: 'dedup_key'
        query_on_bulk_errors: true
        query_duration: PT30S
        async_limit: 5000

2. Index a document successfully.
3. Send the same document again (same source + articleId) so that query_lookup finds an existing match.
4. Observe the NPE repeating every ~20 seconds in logs with no recovery.

Stack trace:

java.lang.NullPointerException: Cannot invoke "java.util.Map.get(Object)" because "bulkOperationsForIndex" is null
    at org.opensearch.dataprepper.plugins.sink.opensearch.index.ExistingDocumentQueryManager.lambda$dropAndReleaseFoundEvents$10(ExistingDocumentQueryManager.java:273)
    at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
    at java.base/java.util.Collections$UnmodifiableCollection.forEach(Collections.java:1093)
    at org.opensearch.dataprepper.plugins.sink.opensearch.index.ExistingDocumentQueryManager.lambda$dropAndReleaseFoundEvents$11(ExistingDocumentQueryManager.java:265)
    at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
    at java.base/java.util.Collections$UnmodifiableCollection.forEach(Collections.java:1093)
    at org.opensearch.dataprepper.plugins.sink.opensearch.index.ExistingDocumentQueryManager.dropAndReleaseFoundEvents(ExistingDocumentQueryManager.java:261)
    at org.opensearch.dataprepper.plugins.sink.opensearch.index.ExistingDocumentQueryManager.runQueryLoop(ExistingDocumentQueryManager.java:135)
    at io.micrometer.core.instrument.composite.CompositeTimer.record(CompositeTimer.java:141)
    at org.opensearch.dataprepper.plugins.sink.opensearch.index.ExistingDocumentQueryManager.run(ExistingDocumentQueryManager.java:113)

Expected behavior
When query_lookup finds an existing document, the duplicate should be silently dropped and the remaining new documents should be released for indexing without throwing an exception. If an exception does occur, the retry mechanism should either successfully recover or route the affected documents to the DLQ — not loop indefinitely.

Environment (please complete the following information):
Data Prepper version: 2.14.1
OpenSearch version: 3.2.0
Deployment: Docker (opensearchproject/data-prepper:2.14.1)
OS: Amazon Linux 2023 (container base image)
Source: kafka_2.13-4.1.1
Index type: custom alias with ISM daily rollover

Actual behavior
Two distinct failure modes observed:
First occurrence — NPE thrown, log says "retrying", suggesting recovery. Documents remain in buffer.
All subsequent cycles — The same NPE fires every ~20 seconds on the same stuck documents. They are never indexed, never dropped, never sent to DLQ. The only resolution is a full pipeline restart, which discards the buffered events entirely.
This makes query_lookup effectively unusable in any scenario where duplicates actually exist, which is precisely its intended use case.

Additional context
The bulkOperationsForIndex map appears to be null or already cleared by the time dropAndReleaseFoundEvents runs, suggesting a possible race condition or lifecycle ordering issue between addBulkOperation and dropAndReleaseFoundEvents — specifically when the lookup returns matches rather than misses.
Non-duplicate documents (where query_lookup finds no existing match) process correctly with no errors. The bug is exclusively triggered when a match is found and the drop logic executes.
Workaround: remove query_lookup from the pipeline configuration entirely. Query-time deduplication can be handled at the search layer using OpenSearch field collapsing on a pre-computed keyword field instead.

Start in ExistingDocumentQueryManager.java, especially runQueryLoop and dropAndReleaseFoundEvents around lines 135 and 261-273, and trace how bulkOperationsForIndex is populated and cleared. Reproduce the query_lookup duplicate-document case from the issue using Data Prepper 2.14.1 and inspect the retry path. Done means duplicates are dropped, new documents are released, and failures do not leave documents stuck in an indefinite NPE loop.