Small updates of TopicRetryableStream & added LazyExecutor

[alex268]

No description provided.

[codecov]

Codecov Report

❌ Patch coverage is 84.61538% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 73.13%. Comparing base (98aab78) to head (84833fe).

Files with missing lines	Patch %	Lines
...in/java/tech/ydb/topic/read/impl/LazyExecutor.java	73.68%	6 Missing and 4 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #721      +/-   ##
============================================
+ Coverage     73.11%   73.13%   +0.01%     
- Complexity     3566     3581      +15     
============================================
  Files           391      392       +1     
  Lines         16497    16548      +51     
  Branches       1730     1736       +6     
============================================
+ Hits          12062    12102      +40     
- Misses         3817     3823       +6     
- Partials        618      623       +5     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[KirillKurdyukov]

Requesting changes for three reproducible issues:

1. Late completion of the old stream can detach the replacement retry stream (TopicRetryableStream.java:57). After fail() removes stream A and starts stream B, A may complete asynchronously. Its callback then calls realStream.getAndSet(null), removes B, and handles A's status as if it belonged to B. In the reproduced scenario this sets isClosed and all subsequent send() calls are dropped. The callback should clear only its captured stream, e.g. with compareAndSet(stream, null), and pass that same stream to onStreamStop.

2. LazyExecutor never grows beyond one worker (LazyExecutor.java:46). Because the LinkedBlockingQueue is unbounded, ThreadPoolExecutor always queues after the single core thread and never reaches MAX_EXECUTOR_THREADS_COUNT = 4. A blocking task therefore stalls every later task. Please either use four core workers for a fixed pool or a bounded/direct-handoff queue for an elastic pool.

3. A retry scheduling failure leaves the stream logically open (TopicRetryableStream.java:146). When scheduler.schedule(...) throws, onClose is called but isClosed remains false. As a result isClosed() reports the wrong state and start() is still allowed after terminal closure. Please set the terminal state before invoking onClose.

The existing tests pass, but I reproduced all three cases with focused regression tests; each currently fails.