added retryable task test && refactored util module

Author: wieceslaw

coordination/src/main/java/tech/ydb/coordination/recipes/util/Listenable.java

@@ -3,15 +3,33 @@
 import java.util.concurrent.ExecutorService;
 import java.util.function.Consumer;
 
+/**
+ * Generic interface for objects that allow adding and removing listeners for events of type T.
+ *
+ * @param <T> the type of event data that listeners will receive
+ */
 public interface Listenable<T> {
+    /**
+     * Adds a listener that will be notified synchronously when the event occurs.
+     *
+     * @param listener the listener to add, must not be null
+     * @throws NullPointerException if listener is null
+     */
     void addListener(Consumer<T> listener);
 
     /**
-     * Listener call will be processed in executor
-     * @param listener
-     * @param executor
+     * Adds a listener that will be notified asynchronously using the provided executor.
+     *
+     * @param listener the listener to add, must not be null
+     * @param executor the executor to use for asynchronous notification, must not be null
+     * @throws NullPointerException if listener or executor is null
      */
     void addListener(Consumer<T> listener, ExecutorService executor);
 
+    /**
+     * Removes the specified listener.
+     *
+     * @param listener the listener to remove
+     */
     void removeListener(Consumer<T> listener);
 }

coordination/src/main/java/tech/ydb/coordination/recipes/util/ListenableContainer.java

@@ -1,54 +1,87 @@
 package tech.ydb.coordination.recipes.util;
 
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Consumer;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * Thread-safe container for managing and notifying listeners.
+ *
+ * @param <T> the type of event data that listeners will receive
+ */
 public class ListenableContainer<T> implements Listenable<T> {
     private static final Logger logger = LoggerFactory.getLogger(ListenableContainer.class);
 
+    // Maps original listeners to potentially wrapped listeners
     private final Map<Consumer<T>, Consumer<T>> listenersMapping = new ConcurrentHashMap<>();
 
+    /**
+     * Notifies all registered listeners with the provided data.
+     * Exceptions thrown by listeners are caught and logged.
+     *
+     * @param data the data to send to listeners
+     * @throws NullPointerException if data is null
+     */
     public void notifyListeners(T data) {
+        Objects.requireNonNull(data, "Data cannot be null");
+
         listenersMapping.values().forEach(listener -> {
             try {
                 listener.accept(data);
             } catch (Exception ex) {
-                logger.error("Listener threw exception", ex);
+                logger.error("Listener threw exception during notification", ex);
             }
         });
     }
 
     @Override
-    public void addListener(Consumer listener) {
+    public void addListener(Consumer<T> listener) {
+        Objects.requireNonNull(listener, "Listener cannot be null");
+
         if (listenersMapping.containsKey(listener)) {
+            logger.debug("Listener already registered, skipping");
             return;
         }
 
         listenersMapping.put(listener, listener);
     }
 
     @Override
-    public void addListener(Consumer listener, ExecutorService executor) {
+    public void addListener(Consumer<T> listener, ExecutorService executor) {
+        Objects.requireNonNull(listener, "Listener cannot be null");
+        Objects.requireNonNull(executor, "Executor cannot be null");
+
         if (listenersMapping.containsKey(listener)) {
+            logger.debug("Listener already registered, skipping");
             return;
         }
 
-        Consumer<T> wrapper = new Consumer<T>() {
-            @Override
-            public void accept(T data) {
-                executor.submit(() -> listener.accept(data));
+        Consumer<T> wrapper = data -> {
+            try {
+                executor.submit(() -> {
+                    try {
+                        listener.accept(data);
+                    } catch (Exception ex) {
+                        logger.error("Asynchronous listener threw exception", ex);
+                    }
+                });
+            } catch (Exception ex) {
+                logger.error("Failed to submit listener task to executor", ex);
             }
         };
+
         listenersMapping.put(listener, wrapper);
     }
 
     @Override
-    public void removeListener(Consumer listener) {
+    public void removeListener(Consumer<T> listener) {
+        Objects.requireNonNull(listener, "Listener cannot be null");
+
         listenersMapping.remove(listener);
     }
 }

coordination/src/main/java/tech/ydb/coordination/recipes/util/RetryableTask.java

@@ -40,7 +40,7 @@ public CompletableFuture<Status> execute() {
         return result;
     }
 
-    private void attemptTask(CompletableFuture<Status> result) {
+    void attemptTask(CompletableFuture<Status> result) {
         try {
             taskSupplier.get().whenComplete((status, throwable) -> {
                 if (throwable != null) {

coordination/src/main/java/tech/ydb/coordination/recipes/util/SessionListenableProvider.java

@@ -2,6 +2,14 @@
 
 import tech.ydb.coordination.CoordinationSession;
 
+/**
+ * Provides access to a Listenable for session state changes.
+ */
 public interface SessionListenableProvider {
+    /**
+     * Gets the Listenable for session state changes.
+     *
+     * @return the Listenable instance for session state changes, never null
+     */
     Listenable<CoordinationSession.State> getSessionListenable();
 }

coordination/src/test/java/tech/ydb/coordination/recipes/util/RetryableTaskTest.java

@@ -0,0 +1,137 @@
+package tech.ydb.coordination.recipes.util;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import tech.ydb.common.retry.RetryPolicy;
+import tech.ydb.core.Status;
+import tech.ydb.core.StatusCode;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+@RunWith(MockitoJUnitRunner.class)
+public class RetryableTaskTest {
+    @Mock
+    private Supplier<CompletableFuture<Status>> taskSupplier;
+
+    @Mock
+    private ScheduledExecutorService executor;
+
+    @Mock
+    private RetryPolicy retryPolicy;
+
+    private RetryableTask retryableTask;
+    private final String taskName = "testTask";
+
+    @Before
+    public void setUp() {
+        retryableTask = new RetryableTask(taskName, taskSupplier, executor, retryPolicy);
+    }
+
+    @Test
+    public void testExecute_SuccessOnFirstAttempt() {
+        Status successStatus = Status.SUCCESS;
+        CompletableFuture<Status> future = CompletableFuture.completedFuture(successStatus);
+
+        when(taskSupplier.get()).thenReturn(future);
+
+        CompletableFuture<Status> result = retryableTask.execute();
+
+        assertTrue(result.isDone());
+        assertEquals(successStatus, result.join());
+    }
+
+    @Test
+    public void testExecute_FailureWithRetries() {
+        Status failureStatus = Status.of(StatusCode.CLIENT_INTERNAL_ERROR);
+        RuntimeException exception = new RuntimeException("Operation failed");
+
+        // First attempt fails
+        CompletableFuture<Status> failedFuture = new CompletableFuture<>();
+        failedFuture.completeExceptionally(exception);
+
+        when(taskSupplier.get())
+                .thenReturn(failedFuture)
+                .thenReturn(CompletableFuture.completedFuture(failureStatus));
+
+        when(retryPolicy.nextRetryMs(anyInt(), anyLong()))
+                .thenReturn(100L) // First retry after 100ms
+                .thenReturn(-1L); // No more retries
+
+        CompletableFuture<Status> result = retryableTask.execute();
+
+        // Verify retry was scheduled
+        verify(executor).schedule(any(Runnable.class), eq(100L), eq(TimeUnit.MILLISECONDS));
+
+        // Simulate retry execution
+        retryableTask.attemptTask(result);
+
+        assertTrue(result.isDone());
+        assertTrue(result.isCompletedExceptionally());
+    }
+
+    @Test
+    public void testExecute_SuccessAfterRetry() {
+        Status successStatus = Status.SUCCESS;
+        RuntimeException exception = new RuntimeException("Temporary failure");
+
+        // First attempt fails
+        CompletableFuture<Status> failedFuture = new CompletableFuture<>();
+        failedFuture.completeExceptionally(exception);
+
+        when(taskSupplier.get())
+                .thenReturn(failedFuture)
+                .thenReturn(CompletableFuture.completedFuture(successStatus));
+
+        when(retryPolicy.nextRetryMs(anyInt(), anyLong()))
+                .thenReturn(0L); // Immediate retry
+
+        CompletableFuture<Status> result = retryableTask.execute();
+
+        // Verify immediate retry was scheduled
+        verify(executor).execute(any(Runnable.class));
+
+        // Simulate retry execution
+        retryableTask.attemptTask(result);
+
+        assertTrue(result.isDone());
+        assertEquals(successStatus, result.join());
+    }
+
+    @Test
+    public void testExecute_NoMoreRetries() {
+        RuntimeException exception = new RuntimeException("Permanent failure");
+
+        CompletableFuture<Status> failedFuture = new CompletableFuture<>();
+        failedFuture.completeExceptionally(exception);
+
+        when(taskSupplier.get()).thenReturn(failedFuture);
+        when(retryPolicy.nextRetryMs(anyInt(), anyLong())).thenReturn(-1L); // No more retries
+
+        CompletableFuture<Status> result = retryableTask.execute();
+
+        assertTrue(result.isDone());
+        assertTrue(result.isCompletedExceptionally());
+    }
+
+    @Test
+    public void testExecute_TaskSupplierThrowsException() {
+        RuntimeException exception = new RuntimeException("Supplier failure");
+
+        when(taskSupplier.get()).thenThrow(exception);
+        when(retryPolicy.nextRetryMs(anyInt(), anyLong())).thenReturn(-1L); // No more retries
+
+        CompletableFuture<Status> result = retryableTask.execute();
+
+        assertTrue(result.isDone());
+        assertTrue(result.isCompletedExceptionally());
+    }
+}