refactor: Removed superfluous AwaitTimeoutException

subprojects/spring-batch-redis-core/src/main/java/com/redis/spring/batch/RedisItemReader.java

@@ -6,6 +6,7 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -34,7 +35,6 @@
 import com.redis.spring.batch.reader.MemKeyValue;
 import com.redis.spring.batch.reader.MemKeyValueRead;
 import com.redis.spring.batch.util.Await;
-import com.redis.spring.batch.util.AwaitTimeoutException;
 import com.redis.spring.batch.util.BatchUtils;
 
 import io.lettuce.core.AbstractRedisClient;
@@ -123,7 +123,7 @@ protected synchronized void doOpen() throws Exception {
 			} catch (InterruptedException e) {
 				Thread.currentThread().interrupt();
 				throw e;
-			} catch (AwaitTimeoutException e) {
+			} catch (TimeoutException e) {
 				List<Throwable> exceptions = jobExecution.getAllFailureExceptions();
 				if (!CollectionUtils.isEmpty(exceptions)) {
 					throw new JobExecutionException("Job execution unsuccessful", exceptions.get(0));
@@ -205,7 +205,7 @@ private StatefulRedisModulesConnection<K, V> connection() {
 	}
 
 	@Override
-	protected synchronized void doClose() throws InterruptedException {
+	protected synchronized void doClose() throws TimeoutException, InterruptedException {
 		if (!queue.isEmpty()) {
 			log.warn(String.format("%s queue still contains %,d elements", getName(), queue.size()));
 		}

subprojects/spring-batch-redis-test/src/main/java/com/redis/spring/batch/test/AbstractTestBase.java

@@ -10,6 +10,7 @@
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeoutException;
 import java.util.function.BooleanSupplier;
 
 import org.junit.jupiter.api.AfterAll;
@@ -135,7 +136,7 @@ void teardown() {
 	}
 
 	@BeforeEach
-	void flushAll() throws InterruptedException {
+	void flushAll() throws TimeoutException, InterruptedException {
 		redisCommands.flushall();
 		awaitUntilNoSubscribers();
 	}
@@ -212,10 +213,12 @@ protected void awaitUntilSubscribers() {
 		} catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
 			throw new ItemStreamException("Interrupted", e);
+		} catch (TimeoutException e) {
+			throw new ItemStreamException("Timeout while waiting for subscribers", e);
 		}
 	}
 
-	protected void awaitUntilNoSubscribers() throws InterruptedException {
+	protected void awaitUntilNoSubscribers() throws TimeoutException, InterruptedException {
 		awaitUntil(() -> redisCommands.pubsubNumpat() == 0);
 	}
 
@@ -289,19 +292,19 @@ public static AbstractRedisClient client(RedisServer server) {
 		return RedisModulesClient.create(server.getRedisURI());
 	}
 
-	public void awaitRunning(JobExecution jobExecution) throws InterruptedException {
+	public void awaitRunning(JobExecution jobExecution) throws TimeoutException, InterruptedException {
 		awaitUntil(jobExecution::isRunning);
 	}
 
-	public void awaitTermination(JobExecution jobExecution) throws InterruptedException {
+	public void awaitTermination(JobExecution jobExecution) throws TimeoutException, InterruptedException {
 		awaitUntilFalse(jobExecution::isRunning);
 	}
 
-	protected void awaitUntilFalse(BooleanSupplier evaluator) throws InterruptedException {
+	protected void awaitUntilFalse(BooleanSupplier evaluator) throws TimeoutException, InterruptedException {
 		awaitUntil(() -> !evaluator.getAsBoolean());
 	}
 
-	protected void awaitUntil(BooleanSupplier evaluator) throws InterruptedException {
+	protected void awaitUntil(BooleanSupplier evaluator) throws TimeoutException, InterruptedException {
 		Await.await().initialDelay(pollDelay).delay(awaitPollInterval).timeout(awaitTimeout).until(evaluator);
 	}
 
@@ -317,50 +320,50 @@ protected void generateAsync(TestInfo info, GeneratorItemReader reader) {
 			} catch (InterruptedException e) {
 				Thread.currentThread().interrupt();
 				throw new ItemStreamException("Data gen interrupted", e);
-			} catch (JobExecutionException e) {
+			} catch (Exception e) {
 				throw new ItemStreamException("Could not run data gen", e);
 			}
 		});
 	}
 
 	protected void generate(TestInfo info, GeneratorItemReader reader)
-			throws JobExecutionException, InterruptedException {
+			throws JobExecutionException, TimeoutException, InterruptedException {
 		generate(info, redisClient, reader);
 	}
 
 	protected void generate(TestInfo info, AbstractRedisClient client, GeneratorItemReader reader)
-			throws JobExecutionException, InterruptedException {
+			throws JobExecutionException, TimeoutException, InterruptedException {
 		TestInfo testInfo = testInfo(info, "generate");
 		RedisItemWriter<String, String, KeyValue<String, Object>> writer = RedisItemWriter.struct(StringCodec.UTF8);
 		writer.setClient(client);
 		run(testInfo, reader, writer);
 	}
 
 	protected void run(TestInfo info, GeneratorItemReader reader, ItemWriter<KeyValue<String, Object>> writer)
-			throws JobExecutionException, InterruptedException {
+			throws JobExecutionException, TimeoutException, InterruptedException {
 		run(info, reader, genItemProcessor, writer);
 	}
 
 	protected <T> JobExecution run(TestInfo info, ItemReader<? extends T> reader, ItemWriter<T> writer)
-			throws JobExecutionException, InterruptedException {
+			throws JobExecutionException, TimeoutException, InterruptedException {
 		return run(info, reader, null, writer);
 	}
 
 	protected <I, O> JobExecution run(TestInfo info, ItemReader<I> reader, ItemProcessor<I, O> processor,
-			ItemWriter<O> writer) throws JobExecutionException, InterruptedException {
+			ItemWriter<O> writer) throws JobExecutionException, TimeoutException, InterruptedException {
 		return run(info, step(info, reader, processor, writer));
 	}
 
 	protected <I, O> JobExecution run(TestInfo info, SimpleStepBuilder<I, O> step)
-			throws JobExecutionException, InterruptedException {
+			throws JobExecutionException, TimeoutException, InterruptedException {
 		return run(job(info).start(faultTolerant(step).build()).build());
 	}
 
 	protected <I, O> FaultTolerantStepBuilder<I, O> faultTolerant(SimpleStepBuilder<I, O> step) {
 		return step.faultTolerant().retryPolicy(new MaxAttemptsRetryPolicy()).retry(RedisCommandTimeoutException.class);
 	}
 
-	protected JobExecution run(Job job) throws JobExecutionException, InterruptedException {
+	protected JobExecution run(Job job) throws JobExecutionException, TimeoutException, InterruptedException {
 		JobExecution execution = jobFactory.run(job);
 		awaitUntilFalse(execution::isRunning);
 		return execution;
@@ -375,7 +378,8 @@ protected <I, S extends I, O> FlushingStepBuilder<S, O> flushingStep(TestInfo in
 		return new FlushingStepBuilder<>(step(info, reader, writer)).idleTimeout(idleTimeout);
 	}
 
-	protected void generateStreams(TestInfo info, int messageCount) throws JobExecutionException, InterruptedException {
+	protected void generateStreams(TestInfo info, int messageCount)
+			throws JobExecutionException, TimeoutException, InterruptedException {
 		GeneratorItemReader gen = generator(3, Item.Type.STREAM);
 		StreamOptions streamOptions = new StreamOptions();
 		streamOptions.setMessageCount(Range.of(messageCount));

subprojects/spring-batch-redis-test/src/test/java/com/redis/spring/batch/test/BatchTests.java

@@ -333,7 +333,7 @@ void readStruct(TestInfo info) throws Exception {
 	}
 
 	@Test
-	void readStreamAutoAck(TestInfo info) throws InterruptedException {
+	void readStreamAutoAck(TestInfo info) throws Exception {
 		String stream = "stream1";
 		String consumerGroup = "batchtests-readStreamAutoAck";
 		Consumer<String> consumer = Consumer.from(consumerGroup, "consumer1");
@@ -390,7 +390,7 @@ void readStreamManualAck(TestInfo info) throws Exception {
 	}
 
 	@Test
-	void readStreamManualAckRecover(TestInfo info) throws InterruptedException {
+	void readStreamManualAckRecover(TestInfo info) throws Exception {
 		String stream = "stream1";
 		Consumer<String> consumer = Consumer.from("batchtests-readStreamManualAckRecover", "consumer1");
 		final StreamItemReader<String, String> reader = streamReader(info, stream, consumer);
@@ -426,7 +426,7 @@ void readStreamManualAckRecover(TestInfo info) throws InterruptedException {
 	}
 
 	@Test
-	void readStreamManualAckRecoverUncommitted(TestInfo info) throws InterruptedException {
+	void readStreamManualAckRecoverUncommitted(TestInfo info) throws Exception {
 		String stream = "stream1";
 		String consumerGroup = "batchtests-readStreamManualAckRecoverUncommitted";
 		Consumer<String> consumer = Consumer.from(consumerGroup, "consumer1");
@@ -506,7 +506,7 @@ void readStreamManualAckRecoverFromOffset(TestInfo info) throws Exception {
 	}
 
 	@Test
-	void readStreamRecoverManualAckToAutoAck(TestInfo info) throws InterruptedException {
+	void readStreamRecoverManualAckToAutoAck(TestInfo info) throws Exception {
 		String stream = "stream1";
 		String consumerGroup = "readStreamRecoverManualAckToAutoAck";
 		Consumer<String> consumer = Consumer.from(consumerGroup, "consumer1");

subprojects/spring-batch-redis-util/src/main/java/com/redis/spring/batch/util/Await.java

@@ -4,6 +4,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.BooleanSupplier;
 
 public class Await {
@@ -35,11 +36,11 @@ public Await timeout(Duration timeout) {
 	 * Blocks until test is true
 	 *
 	 * @param test boolean supplier to wait for
-	 * @throws InterruptedException  if interrupted while waiting
-	 * @throws AwaitTimeoutException if condition was not fulfilled within timeout
-	 *                               duration
+	 * @throws InterruptedException if interrupted while waiting
+	 * @throws TimeoutException     if condition not fulfilled within timeout
+	 *                              duration
 	 */
-	public void until(BooleanSupplier test) throws InterruptedException {
+	public void until(BooleanSupplier test) throws TimeoutException, InterruptedException {
 		ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
 		try {
 			executor.scheduleWithFixedDelay(() -> {
@@ -49,14 +50,14 @@ public void until(BooleanSupplier test) throws InterruptedException {
 			}, initialDelay.toMillis(), delay.toMillis(), TimeUnit.MILLISECONDS);
 			boolean terminated = executor.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS);
 			if (!terminated) {
-				throw new AwaitTimeoutException(String.format("Condition not fulfilled within %s", timeout));
+				throw new TimeoutException(String.format("Condition not fulfilled within %s", timeout));
 			}
 		} finally {
 			executor.shutdown();
 		}
 	}
 
-	public void untilFalse(BooleanSupplier test) throws InterruptedException {
+	public void untilFalse(BooleanSupplier test) throws TimeoutException, InterruptedException {
 		until(() -> !test.getAsBoolean());
 	}