ykisana opened a new pull request, #20821: URL: https://github.com/apache/druid/pull/20821
Fixes #20817. ### Description `KafkaMultiSupervisorTest.test_ingestIntoSingleDatasource_fromDifferentTopics` didn't fail because ingestion was slow. It failed because the Overlord's `assignTask` request to the Indexer hung for 60 seconds. The hang is caused by a race in the shared `AsyncRequest` helper added in #20784, which leaves an HTTP connection that the server never reads from again. #### Fixed a race between an async response and its timeout The race happens when the future of a long-polling request completes on another thread just as Jetty's async timeout fires: 1. The future's thread wins `AsyncRequest`'s at-most-once guard and writes the response. Jetty then rejects its `AsyncContext.complete()`, because while a request is timing out, Jetty only allows it to be completed from the thread running the `onTimeout` listeners (`HttpChannelState.complete()`). 2. The timeout listener sees that the request was already completed and returns without completing it. 3. Jetty calls `sendError(500)` on the committed response, which throws `IllegalStateException: COMMITTED`, and the exchange is never finished. 4. The client received the whole response, so it reuses the keep-alive connection. Jetty never reads from that connection again, and the next request sent on it hangs until the client's read timeout. In the [failing CI run](https://github.com/apache/druid/actions/runs/41264781389/job/124725198050), this happened on the Indexer's worker-sync endpoint (`TaskManagementResource`), which the embedded tests poll with a 1s timeout: ``` 02:39:40,131 WARN [SegmentChangeRequestHistory] AsyncRequest - Request timed out or is closed already (java.lang.IllegalStateException: s=WOKEN rs=EXPIRE os=COMMITTED ...) 02:39:40,132 WARN [qtp...] QueuedThreadPool - Job failed java.lang.IllegalStateException: COMMITTED at HttpChannelState.sendError(HttpChannelState.java:810) ... 02:39:40,197 INFO HttpRemoteTaskRunner - Assigning task [index_kafka_..._kchhhigg] to worker [10.1.0.223:8091] 02:40:40,258 WARN RetryUtils - Retrying (1 of 2) ... Request to assign task[index_kafka_..._kchhhigg] ... failed ... Caused by: io.netty.handler.timeout.ReadTimeoutException ``` The `assignTask` request was sent on the stuck connection and hung for the 60s `assignRequestHttpTimeout`. `HttpRemoteTaskRunner` has one pending-task thread and skips workers that have an unacknowledged assignment, so no task was assigned for those 60 seconds, and the test's 60s wait ran out. Before #20784, the timeout listeners in `TaskManagementResource` and `SegmentListerResource` always completed the request. `OverlordResource` and `MessageRelayResource` completed it only if they won the guard, so they already had this latent issue. This change fixes all `AsyncRequest` callers. The fix is in `AsyncRequest`: - The timeout listener now always calls `event.getAsyncContext().complete()` in a `finally` after its guarded completion. Jetty allows this on the `onTimeout` thread, and it does nothing if the request is already complete. - `complete()` is now `synchronized`, so if the future's thread is still writing its response, the timeout listener waits for the write to finish instead of cutting the response off. Concurrency notes, per the concurrency checklist: - The lock is the per-request `AsyncRequest` instance. Only `complete()` takes it, and `completed` is `@GuardedBy("this")`. - `complete()` calls the caller's `onSuccess`/`onFailure` function and writes the response while holding the lock. This is intentional: it is what makes the timeout listener wait for a write in progress. Only that request's timeout listener contends for the lock. The Javadocs of `onSuccess`/`onFailure` now say they must not block. - There is no lock-ordering issue with Jetty, which runs `onTimeout` listeners after releasing its `HttpChannelState` lock. `future.cancel(true)` in the listener can re-enter `complete()` through `onFailure` on the same thread, which the reentrant monitor allows. Alternatives considered: - Raising the test's wait timeout, as the issue suggests, would hide the problem rather than fix it. - Reverting `TaskManagementResource` to its own `AsyncContext` handling would leave the other callers affected. Not changed in this PR: - `multi-stage-query`'s `WorkerResource` (channel-data long poll) has a similar pattern. Its timeout listener completes the request only if it wins `responseResolved`, and `onSuccess` calls `asyncContext.complete()` in a `finally` without handling the exception. This is worth a follow-up. - When the future's thread loses the race described above, it still logs `Request timed out or is closed already` at WARN, as `OverlordResource` did before #20784. No release note is needed: the regression from #20784 has not been released. #### Tests - `AsyncRequestTest`: two new unit tests. - `test_start_completesRequestOnTimeout_whenFutureRespondedButCouldNotComplete`: the mock `AsyncContext` rejects `complete()` the way Jetty does, and the test asserts that the timeout listener still completes the request without changing the response. - `test_start_timeoutWaitsForResponseBeingWrittenByAnotherThread`: asserts that the timeout listener blocks while another thread is writing the response. - `AsyncRequestJettyTest` (new): reproduces the CI failure against a real Jetty server. It uses Druid's `HttpClient` with a single connection, following `DirectDruidClientAbortHttpTest`. Latches force the order seen in CI, so the test does not depend on timing. It asserts that a second request on the same connection is answered. Verification (in a checkout of this branch): ``` # tests, with checkstyle, PMD, forbidden-apis and strict compilation enabled; spotbugs on the changed package mvn -B test spotbugs:check -pl server,indexing-service -am -DstrictCompile -Dspotbugs.onlyAnalyze='org.apache.druid.server.http.-' \ -Dtest="org.apache.druid.server.http.AsyncRequestTest,org.apache.druid.server.http.AsyncRequestJettyTest,org.apache.druid.indexing.overlord.http.OverlordResourceTest" \ -Dsurefire.failIfNoSpecifiedTests=false -Dweb.console.skip=true mvn -B -pl server apache-rat:check -Prat ``` - With the fix: `AsyncRequestTest` 16/16, `AsyncRequestJettyTest` 1/1, `OverlordResourceTest` 52/52. Static checks pass. - Without the fix in `AsyncRequest.java` (new tests kept): both new `AsyncRequestTest` tests fail. `AsyncRequestJettyTest` fails with a `ReadTimeoutException` on the second request, and Jetty logs the same `IllegalStateException: COMMITTED` from `HttpChannelState.sendError(HttpChannelState.java:810)` as in CI. - `KafkaMultiSupervisorTest` (embedded) passed 2/2 with an earlier revision of the fix that had the same logic, before the `finally` and the Javadoc notes were added. The race did not occur in that run, so this only shows that nothing regressed. <hr> ##### Key changed/added classes in this PR * `AsyncRequest` * `AsyncRequestTest` * `AsyncRequestJettyTest` <hr> This PR has: - [x] been self-reviewed. - [x] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.) - [x] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader. - [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met. #### Generative AI usage Generated by: Claude Opus 5.5 Root cause, fix, tests, PR description by Claude. All code manually reviewed by @ykisana. -- 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]
