Caideyipi opened a new pull request, #18222:
URL: https://github.com/apache/iotdb/pull/18222
## Description
The async Pipe sink retry queue could continuously drain and resubmit events
when a receiver remained under memory pressure. A receiver returning
`PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION` or `WRITE_PROCESS_REJECT`
(including either code in a sub-status) therefore caused sustained retry
traffic and high sender CPU usage.
This PR bounds the duration of normal retries and changes prolonged receiver
unavailability into a low-frequency probe mode. Retry events are retained
throughout the transition; this change throttles retries and does not introduce
a retry TTL or data loss policy.
### Retry behavior
The behavior is tracked independently for each receiver endpoint:
1. The first temporary-unavailable response activates the endpoint backoff
and records the start time.
2. During the regular retry phase, callers reserve retry times serially for
that endpoint. The delay uses the existing exponential backoff bounded by
`pipe_sink_subtask_sleep_interval_init_ms` and
`pipe_sink_subtask_sleep_interval_max_ms`. Serial reservations prevent several
queued events from waking and resubmitting to the same receiver at once.
3. After the endpoint has remained unavailable for
`pipe_async_sink_retry_max_duration_ms`, regular retries stop. One caller may
send a probe immediately, and at most one further probe is allowed per
`pipe_async_sink_retry_probe_interval_ms`.
4. Other events remain in the retry queue while the probe slot is cooling
down. The sink publishes the remaining cooldown through
`PipeSinkWithSchedulingDelay`, so the subtask is delayed instead of repeatedly
draining the queue.
5. A successful receiver response clears the endpoint state and cancels an
unconsumed scheduling delay when no endpoint remains unavailable. The success
update uses `ConcurrentHashMap.computeIfPresent` and checks the active failure
window so that an older in-flight success cannot erase a newer failure state.
Backoff/probe ownership is endpoint-level, while the existing
scheduling-delay API is sink/subtask-level. Consequently, if any endpoint is in
probe cooldown, retry draining for that sink is delayed. This intentionally
applies backpressure to the shared retry queue and prioritizes stopping an
overload retry storm over continuing maximum throughput to other endpoints
during the cooldown.
### Hidden configuration
Two hidden system properties control the transition:
| Property | Default | Semantics |
| --- | ---: | --- |
| `pipe_async_sink_retry_max_duration_ms` | `60000` | Maximum duration of
regular retries after the first temporary-unavailable response. `0` enters
probe mode immediately; a negative value disables the transition and retains
regular retries indefinitely. |
| `pipe_async_sink_retry_probe_interval_ms` | `30000` | Minimum interval
between probe requests after the maximum retry duration. Values below `1` are
clamped to `1` ms. |
For compatibility with the previous connector naming convention, the parser
also accepts `pipe_async_connector_retry_max_duration_ms` and
`pipe_async_connector_retry_probe_interval_ms`. The `pipe_async_sink_*` names
take precedence when both are set.
These properties are intentionally not added to any `.properties`, `.conf`,
or configuration template. They remain internal tuning parameters while the
default behavior is enabled for all async Pipe sinks.
### Async client and error handling
An async client may already have been borrowed when an event discovers that
its endpoint is in probe cooldown. In that case the handler now returns the
client to the pool before routing the event through `onError`, which safely
puts the event back into the retry queue. The sliced-request fallback path
follows the same ownership rule.
The cooldown uses `PipeRuntimeSinkNonReportTimeConfigurableException` as an
internal scheduling signal. `PipeSinkSubtask` preserves this exception for
heartbeat events instead of wrapping it as a `PipeConnectionException`; this
avoids triggering a meaningless handshake while the receiver is deliberately
cooling down.
### Concurrency and edge cases
- Mutable endpoint backoff state is guarded by synchronization on the
endpoint state object.
- Probe acquisition and next-probe reservation are atomic, so concurrent
callbacks cannot obtain the same probe slot.
- Endpoint removal is atomic with respect to concurrent failure/success
callbacks.
- Wait loops periodically observe sink closure and endpoint recovery, and
interrupted waits preserve the interrupt status.
- Time addition saturates at `Long.MAX_VALUE` to avoid overflow for large
hidden-property values.
- Closing the sink retains the existing event reference-count cleanup
behavior.
- Retry events are never discarded because the maximum duration is exceeded.
### Verification
- Applied Spotless to `iotdb-core/datanode` and `iotdb-core/node-commons`.
- Compiled the changed node-commons code with both English and Chinese
locale sources.
- Compiled the changed DataNode classes against both English and Chinese
i18n classpaths.
- Ran 12 focused JUnit tests from `PipeTransferTrackableHandlerTest` and
`PipeSinkSubtaskTest`; all passed.
- Added coverage for endpoint retry serialization, transition to a single
probe, recovery from probe mode, async-client return during cooldown, and
heartbeat preservation of the non-report scheduling exception.
- Verified English/Chinese message-key parity and ran `git diff --check`.
The normal DataNode Maven test invocation is currently blocked during
unrelated full-module compilation by missing symbols such as `IFill`,
`Accumulator`, and `TopKRuntimeFilter`. The focused tests were therefore
compiled and run with the Maven-generated test classpath.
<hr>
This PR has:
- [x] been self-reviewed.
- [x] concurrent read
- [x] concurrent write
- [x] concurrent read and write
- [x] added unit tests or modified existing tests to cover new code paths.
<hr>
##### Key changed/added classes (or packages if there are too many classes)
in this PR
- `IoTDBDataRegionAsyncSink`: endpoint retry serialization, retry-duration
tracking, probe admission, scheduling delay, and recovery.
- `PipeTransferTrackableHandler`: async-client ownership and retry-queue
handling when a probe is delayed.
- `PipeSinkSubtask`: preservation of the non-report scheduling signal on the
heartbeat path.
- `CommonConfig`, `PipeConfig`, and `PipeDescriptor`: hidden defaults,
accessors, logging, aliases, and property parsing.
--
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]
