fix!: bound bulk write waits and cancel timed-out streams - #109
Conversation
|
@codex review When you finish, append exactly this marker to the review summary: |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 31d60c8553
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
@codex review When you finish, append exactly this marker to the review summary: |
|
Codex Review: Didn't find any major issues. Keep them coming! Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
fengjiachun
changed the title
- preserve the completed() failure via addSuppressed when the terminal stream failure is rethrown, and assert it in the concurrent-abort test - document that timeoutMsPerMessage bounds each blocking wait independently, so one writeNext() may wait a small multiple of it - document that writeNext() no longer blocks indefinitely on the in-flight limit and becomes terminal on timeout - note that abort() may run on the shared timeout-scheduler thread
c77cda1 to
c500c64
Compare
There was a problem hiding this comment.
Pull request overview
This PR moves bulk-write timeout and cancellation handling into the Java SDK so that bulk-write calls cannot block indefinitely and the underlying Flight DoPut call is actively cancelled on timeout/interruption/early close.
Changes:
- Apply
timeoutMsPerMessageto additional blocking waits (in-flight limiter acquisition, stream readiness, per-message response waits, and final stream completion). - Cancel the underlying Flight call on timeout/interruption/abort, and force gRPC channel shutdown after a graceful shutdown window.
- Add regression tests covering bounded waits and cancellation paths across
BulkWriteClient,BulkWriteService, andBulkFlightClient.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| ingester-protocol/src/test/java/io/greptime/WriteLimitTest.java | Asserts interrupt status is preserved when limiter acquisition is interrupted. |
| ingester-protocol/src/test/java/io/greptime/BulkWriteClientTest.java | Adds tests for limiter timeout behavior and interruption-triggered stream cancellation. |
| ingester-protocol/src/main/java/io/greptime/limit/InFlightLimiter.java | Restores thread interrupt status when semaphore acquisition is interrupted. |
| ingester-protocol/src/main/java/io/greptime/BulkWriteClient.java | Uses a bounded limiter for bulk in-flight slots and aborts/cancels stream on limiter interruption/timeout. |
| ingester-protocol/src/main/java/io/greptime/BulkWrite.java | Updates timeout documentation to reflect broader bulk-write blocking semantics. |
| ingester-protocol/src/main/java/io/greptime/BulkStreamWriter.java | Documents terminal-stream semantics after timeout/interruption/abort. |
| ingester-bulk-protocol/src/test/java/org/apache/arrow/flight/BulkFlightClientTest.java | Adds readiness-timeout cancellation test and verifies forced channel shutdown behavior. |
| ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java | Adds tests for bounded completion waits, abort semantics, and cancellation on close/timeout. |
| ingester-bulk-protocol/src/main/java/org/apache/arrow/flight/BulkFlightClient.java | Adds readiness timeout to DoPut creation, cancels underlying call on readiness timeout/interruption, and enforces forced channel shutdown. |
| ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java | Binds response/completion waits to timeout and cancels DoPut on timeout/abort/close paths. |
| ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java | Wires timeout through to Flight client DoPut creation. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- log the cancel failure in BulkWriteService#close so it is not silently swallowed when the stream already completed - clarify DEFAULT_TIMEOUT_MS_PER_MESSAGE javadoc: it bounds each blocking wait independently, not the end-to-end operation
What changed
timeoutMsPerMessageto bulk limiter acquisition, Flight stream readiness, per-message responses, and final stream completionWhy
flink-connector-greptimedb#21 adds a connector-side executor wrapper because the SDK can otherwise keep a Flink task blocked indefinitely. The wrapper is a useful stopgap, but interrupting the wrapper thread does not guarantee that the SDK's DoPut call has stopped.
This change puts the timeout and cancellation ownership in the SDK. It preserves the existing public
startPutsignature and reuses the existing bulk timeout setting instead of introducing another configuration value.A timed-out stream is terminal and must be closed/recreated by the caller. Automatic replay/recovery and route-resolution timeouts are intentionally outside this PR.
Validation
mvn spotless:checkmvn testBulkWriteService,BulkFlightClient, andBulkWriteClientgit diff --checkBreaking behavior change
writeNext()previously blocked indefinitely whenmaxRequestsInFlightslots were exhausted. It now fails aftertimeoutMsPerMessageand the stream becomes terminal (it must be closed and recreated). Callers that relied on unbounded blocking as flow control should sizetimeoutMsPerMessageaccordingly. Note that the timeout bounds each blocking wait (slot acquisition, stream readiness, per-message response, completion) independently, so a single call may wait up to a small multiple of the configured value in the worst case.