Pipe: throttle prolonged async sink retries

[Caideyipi]

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.

---

This PR has:

• been self-reviewed.
  • concurrent read
  • concurrent write
  • concurrent read and write
• added unit tests or modified existing tests to cover new code paths.

---

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.