Skip to content

Small updates of TopicRetryableStream & added LazyExecutor #721

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alex268 merged 2 commits into from
Sep 10, 2026
Merged

Small updates of TopicRetryableStream & added LazyExecutor #721

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fdf5ef7
Extended TopicRetryableStream: added the current stream processing
alex268 Sep 10, 2026
84833fe
Added LazyExecutor utility class
alex268 Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 40 additions & 20 deletions topic/src/main/java/tech/ydb/topic/impl/TopicRetryableStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,13 @@
import tech.ydb.core.Status;
import tech.ydb.core.StatusCode;

public abstract class TopicRetryableStream<R extends Message, W extends Message> {
public abstract class TopicRetryableStream<R extends Message, W extends Message, S extends TopicStream<R, W>> {
protected final String debugId;
private final Logger logger;
private final RetryConfig retryConfig;
private final ScheduledExecutorService scheduler;

private final AtomicReference<TopicStream<R, W>> realStream = new AtomicReference<>();
private final AtomicReference<S> realStream = new AtomicReference<>();
private final AtomicInteger streamCount = new AtomicInteger(0);
private final RetryState state = new RetryState();

Expand All @@ -32,34 +32,37 @@ public TopicRetryableStream(Logger logger, String debugId, RetryConfig config, S
this.scheduler = scheduler;
}

protected abstract TopicStream<R, W> createNewStream(String debugId);
protected abstract S createNewStream(String debugId);

protected abstract void onNext(R message);
protected abstract void onNext(S stream, R message);

protected abstract void onRetry(Status status);
protected abstract void onClose(Status status);
protected abstract void onRetry(S stream, Status status);
protected abstract void onClose(S stream, Status status);

public void start() {
if (isClosed) {
logger.warn("[{}] double start of closed stream, ignored", debugId);
return;
}

String streamID = debugId + '.' + streamCount.incrementAndGet();
TopicStream<R, W> stream = createNewStream(streamID);
S stream = createNewStream(streamID);

if (!realStream.compareAndSet(null, stream)) {
logger.warn("[{}] double start of stream, skipping", debugId);
return;
}

stream.start(this::onNext).whenComplete((status, th) -> {
realStream.compareAndSet(stream, null);
stream.start(msg -> onNext(stream, msg)).whenComplete((status, th) -> {
if (!realStream.compareAndSet(stream, null)) {
return;
}
if (status != null) {
onStreamStop(status, retryConfig.getStatusRetryPolicy(status));
onStreamStop(stream, status, retryConfig.getStatusRetryPolicy(status));
}
if (th != null) {
Status wrapped = Status.of(StatusCode.CLIENT_INTERNAL_ERROR, th);
onStreamStop(wrapped, retryConfig.getThrowableRetryPolicy(th));
onStreamStop(stream, wrapped, retryConfig.getThrowableRetryPolicy(th));
}
});
}
Expand All @@ -68,8 +71,21 @@ protected void resetRetries() {
state.reset();
}

public boolean isClosed() {
return isClosed;
}

public void fail(Status status) {
S closed = realStream.getAndSet(null);
if (closed != null) {
logger.warn("[{}] failed by application-side error {}", debugId, status);
closed.close();
onStreamStop(closed, status, retryConfig.getStatusRetryPolicy(status));
}
}

public void send(W msg) {
TopicStream<R, W> stream = realStream.get();
S stream = realStream.get();
if (stream == null) {
logger.warn("[{}] send message before stream is ready", debugId);
return;
Expand All @@ -79,51 +95,55 @@ public void send(W msg) {

public boolean close() {
isClosed = true;
TopicStream<R, W> stream = realStream.getAndSet(null);
S stream = realStream.getAndSet(null);
if (stream == null) {
return false;
}

stream.close();
onStreamStop(stream, Status.SUCCESS, null);
return true;
}

private void onStreamStop(Status status, RetryPolicy policy) {
private void onStreamStop(S closed, Status status, RetryPolicy policy) {
if (isClosed) { // stream was already closed (usually with success)
onClose(status);
onClose(closed, status);
return;
}

if (policy == null) {
logger.warn("[{}] stopped by non-retryable status {}", debugId, status);
onClose(status);
isClosed = true;
onClose(closed, status);
return;
}

long nextRetryMs = state.nextRetryMs(policy);

if (nextRetryMs < 0) {
logger.warn("[{}] stopped after retry policy evaluation for status {}", debugId, status);
onClose(status);
isClosed = true;
onClose(closed, status);
return;
}

if (nextRetryMs == 0) { // retry immediately
logger.warn("[{}] retry #{}. Retry immediately...", debugId, state.retryNumber());
onRetry(status);
onRetry(closed, status);
start();
return;
}

// retry scheduling
logger.warn("[{}] retry #{}. Scheduling reconnect in {}ms...", debugId, state.retryNumber(), nextRetryMs);
onRetry(status);
onRetry(closed, status);

try {
scheduler.schedule(this::start, nextRetryMs, TimeUnit.MILLISECONDS);
} catch (Exception ex) {
logger.error("[{}] cannot schedule reconnect, stopping", debugId, ex);
onClose(status);
isClosed = true;
onClose(closed, status);
}
}

Expand Down
82 changes: 82 additions & 0 deletions topic/src/main/java/tech/ydb/topic/read/impl/LazyExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
package tech.ydb.topic.read.impl;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
*
* @author Aleksandr Gorshenin {@literal <alexandr268@ydb.tech>}
*/
public class LazyExecutor implements Executor, AutoCloseable {
private static final Logger logger = LoggerFactory.getLogger(LazyExecutor.class);
private static final int MAX_EXECUTOR_THREADS_COUNT = 4;

private final String name;
private final Executor custom;
private final AtomicInteger threadsCount = new AtomicInteger();
private final AtomicReference<ExecutorService> service = new AtomicReference<>();
private volatile boolean isStopped = false;

public LazyExecutor(String name, Executor custom) {
this.name = name;
this.custom = custom;
}

@Override
public void execute(Runnable command) {
if (isStopped) {
return;
}

if (custom != null) {
custom.execute(command);
return;
}

ExecutorService local = service.get();
while (!isStopped && local == null) {
ThreadFactory factory = r -> new Thread(r, name + "-" + threadsCount.incrementAndGet());
ExecutorService pool = Executors.newFixedThreadPool(MAX_EXECUTOR_THREADS_COUNT, factory);
if (!service.compareAndSet(local, pool)) {
pool.shutdown();
}
local = service.get();
}

if (!isStopped) {
local.execute(command);
} else {
shutdown(service.getAndSet(null));
}
}

@Override
public void close() {
isStopped = true;
shutdown(service.getAndSet(null));
}

private void shutdown(ExecutorService service) {
if (service == null) {
return;
}

try {
service.shutdown();
if (!service.awaitTermination(100, TimeUnit.MILLISECONDS)) {
service.shutdownNow();
}
} catch (InterruptedException e) {
logger.warn("executor {} shutdown interrupted", name, e);
Thread.currentThread().interrupt();
}
}
}
8 changes: 4 additions & 4 deletions topic/src/main/java/tech/ydb/topic/write/impl/WriteSession.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
/**
* @author Nikolay Perfilov
*/
public final class WriteSession extends TopicRetryableStream<FromServer, FromClient> {
public final class WriteSession extends TopicRetryableStream<FromServer, FromClient, WriteSession.Stream> {
public interface Stream extends TopicStream<FromServer, FromClient> { }

private static final Logger logger = LoggerFactory.getLogger(WriteSession.class);
Expand Down Expand Up @@ -108,7 +108,7 @@ WriteAck mapAck(WriteAck.Statistics statistics, YdbTopic.StreamWriteMessage.Writ
}

@Override
public void onRetry(Status status) {
public void onRetry(Stream stream, Status status) {
logger.warn("[{}] Session onRetry with status {} called", debugId, status);
listener.onStop(status);
if (errorsHandler != null) {
Expand All @@ -117,7 +117,7 @@ public void onRetry(Status status) {
}

@Override
public void onClose(Status status) {
public void onClose(Stream stream, Status status) {
logger.info("[{}] Session closed with status {}", debugId, status);
listener.onClose(status);
if (errorsHandler != null && !status.isSuccess()) {
Expand All @@ -126,7 +126,7 @@ public void onClose(Status status) {
}

@Override
public void onNext(YdbTopic.StreamWriteMessage.FromServer message) {
public void onNext(Stream stream, YdbTopic.StreamWriteMessage.FromServer message) {
if (message.hasInitResponse()) {
onInitResponse(message.getInitResponse());
} else if (message.hasWriteResponse()) {
Expand Down
Loading
Loading