[TH2-5204] external executor

Dockerfile

@@ -1,4 +1,4 @@
-FROM gradle:7.6-jdk8 AS build
+FROM gradle:7.6-jdk11 AS build
 ARG release_version
 ARG bintray_user
 ARG bintray_key

README.md

@@ -209,6 +209,9 @@ Test events have mandatory parameters that are verified when storing an event. T
 
 ## Release notes
 
+### 5.4.0
+* Using internal executor instead of ForkJoinPool.commonPool() to process intermediate tasks 
+
 ### 5.3.0
 + Migrated to th2 gradle plugin: `0.0.5` (bom: `4.6.1`)
 + Updated bom: `4.6.1`

cradle-cassandra/build.gradle

@@ -20,12 +20,12 @@ dependencies {
     annotationProcessor "com.datastax.oss:java-driver-mapper-processor"
     // ---
 
-    testImplementation 'org.testng:testng:7.10.1'
-    testImplementation 'org.assertj:assertj-core:3.25.3'
+    testImplementation 'org.testng:testng:7.10.2'
+    testImplementation 'org.assertj:assertj-core:3.26.0'
     testImplementation "org.apache.logging.log4j:log4j-slf4j2-impl"
     testImplementation 'org.apache.logging.log4j:log4j-core'
-    testImplementation 'org.mockito:mockito-core:5.11.0'
-    testImplementation 'org.testcontainers:cassandra:1.19.7'
+    testImplementation 'org.mockito:mockito-core:5.12.0'
+    testImplementation 'org.testcontainers:cassandra:1.19.8'
 }
 
 test {

cradle-cassandra/src/main/java/com/exactpro/cradle/cassandra/CassandraCradleStorage.java

@@ -746,7 +746,9 @@ protected CompletableFuture<CradleResultSet<String>> doGetScopesAsync(BookId boo
 				selectExecutor,
 				readAttrs
 		);
-		return iteratorProvider.nextIterator().thenApplyAsync(it -> new CassandraCradleResultSet<>(it, iteratorProvider));
+		return iteratorProvider
+				.nextIterator()
+				.thenApplyAsync(it -> new CassandraCradleResultSet<>(it, iteratorProvider), composingService);
 	}
 
 	@Override
@@ -966,7 +968,9 @@ protected CompletableFuture<CradleResultSet<String>> doGetSessionAliasesAsync(Bo
 				selectExecutor,
 				readAttrs
 		);
-		return iteratorProvider.nextIterator().thenApplyAsync(it -> new CassandraCradleResultSet<>(it, iteratorProvider));
+		return iteratorProvider
+				.nextIterator()
+				.thenApplyAsync(it -> new CassandraCradleResultSet<>(it, iteratorProvider), composingService);
 	}
 
 	@Override
@@ -1010,7 +1014,9 @@ protected CompletableFuture<CradleResultSet<String>> doGetSessionGroupsAsync(Boo
 				selectExecutor,
 				readAttrs
 		);
-		return iteratorProvider.nextIterator().thenApplyAsync(it -> new CassandraCradleResultSet<>(it, iteratorProvider));
+		return iteratorProvider
+				.nextIterator()
+				.thenApplyAsync(it -> new CassandraCradleResultSet<>(it, iteratorProvider), composingService);
 	}
 
 	@Override

cradle-cassandra/src/main/java/com/exactpro/cradle/cassandra/dao/intervals/CassandraIntervalsWorker.java

@@ -182,7 +182,7 @@ public CompletableFuture<Iterable<Interval>> getIntervalsPerDayAsync(BookId book
                 () -> operator.getIntervals(bookId.getName(), date, fromTime, toTime, crawlerName, crawlerVersion, crawlerType, readAttrs),
                 converter::getEntity,
                 queryInfo)
-                .thenApplyAsync((result) ->  getIntervalsIterator(result, queryInfo));
+                .thenApplyAsync((result) ->  getIntervalsIterator(result, queryInfo), composingService);
     }
 
     @Override
@@ -273,7 +273,7 @@ private CompletableFuture<Iterable<Interval>> getIntervalsPerDayAsync(BookId boo
                         () -> operator.getIntervals(bookId.getName(), date, fromTime, toTime, crawlerName, crawlerVersion, crawlerType, readAttrs),
                         converter::getEntity,
                         queryInfo)
-                .thenApplyAsync((result) ->  getIntervalsIterator(result, queryInfo));
+                .thenApplyAsync((result) ->  getIntervalsIterator(result, queryInfo), composingService);
     }
 
     @Override

cradle-cassandra/src/main/java/com/exactpro/cradle/cassandra/dao/messages/PageGroupsIteratorProvider.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 Exactpro (Exactpro Systems Limited)
+ * Copyright 2023-2024 Exactpro (Exactpro Systems Limited)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -72,6 +72,6 @@ public CompletableFuture<Iterator<String>> nextIterator() {
                     PageGroupEntity::getGroup);
 
             return new UniqueIterator<>(convertingIterator, registry);
-        });
+        }, composingService);
     }
 }

cradle-cassandra/src/main/java/com/exactpro/cradle/cassandra/dao/messages/PageSessionsIteratorProvider.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 Exactpro (Exactpro Systems Limited)
+ * Copyright 2023-2024 Exactpro (Exactpro Systems Limited)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -71,6 +71,6 @@ public CompletableFuture<Iterator<String>> nextIterator() {
                     PageSessionEntity::getSessionAlias);
 
             return new UniqueIterator<>(convertingIterator, registry);
-        });
+        }, composingService);
     }
 }

cradle-cassandra/src/main/java/com/exactpro/cradle/cassandra/dao/testevents/PageScopesIteratorProvider.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 Exactpro (Exactpro Systems Limited)
+ * Copyright 2023-2024 Exactpro (Exactpro Systems Limited)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -69,6 +69,6 @@ public CompletableFuture<Iterator<String>> nextIterator() {
                     PageScopeEntity::getScope);
 
             return new UniqueIterator<>(convertingIterator, registry);
-        });
+        }, composingService);
     }
 }

cradle-cassandra/src/main/java/com/exactpro/cradle/cassandra/retries/SelectQueryExecutor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2022 Exactpro (Exactpro Systems Limited)
+ * Copyright 2021-2024 Exactpro (Exactpro Systems Limited)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -57,7 +57,7 @@ public <T> CompletableFuture<T> executeSingleRowResultQuery(Supplier<Completable
 	{
 		CompletableFuture<T> f = new CompletableFuture<>();
 		query.get()
-				.whenCompleteAsync((result, error) -> onCompleteSingle(result, error, f, mapper, queryInfo, 0));
+				.whenCompleteAsync((result, error) -> onCompleteSingle(result, error, f, mapper, queryInfo, 0), composingService);
 		return f;
 	}
 
@@ -153,7 +153,7 @@ private <T> void onCompleteSingle(T result, Throwable error, CompletableFuture<T
 			CompletableFuture.runAsync(
 							() -> logger.debug("Retrying request ({}) '{}' and CL {} with delay {}ms after error: '{}'",
 									retryCount + 1, queryInfo, stmt.getConsistencyLevel(), delay, error.getMessage()),
-							CompletableFuture.delayedExecutor(delay, TimeUnit.MILLISECONDS))
+							CompletableFuture.delayedExecutor(delay, TimeUnit.MILLISECONDS, composingService))
 					.thenComposeAsync(r -> session.executeAsync(stmt), composingService)
 					.thenApplyAsync(AsyncPagingIterable::one, composingService)
 					.thenApplyAsync(row -> row == null ? null : mapper.apply(row), composingService)
@@ -189,7 +189,7 @@ private <T> void onCompleteMulti(MappedAsyncPagingIterable<T> result, Throwable
 			CompletableFuture.runAsync(
 							() -> logger.debug("Retrying request ({}) '{}' and CL {} with delay {}ms after error: '{}'",
 									retryCount + 1, queryInfo, stmt.getConsistencyLevel(), delay, error.getMessage()),
-							CompletableFuture.delayedExecutor(delay, TimeUnit.MILLISECONDS))
+							CompletableFuture.delayedExecutor(delay, TimeUnit.MILLISECONDS, composingService))
 					.thenComposeAsync(r -> session.executeAsync(stmt), composingService)
 					.thenApplyAsync(row -> row.map(mapper), composingService)
 					.whenCompleteAsync(

cradle-core/build.gradle

@@ -10,8 +10,8 @@ dependencies {
 
     testImplementation 'org.apache.logging.log4j:log4j-slf4j2-impl'
     testImplementation 'org.apache.logging.log4j:log4j-core'
-    testImplementation 'org.testng:testng:7.9.0'
-    testImplementation 'org.assertj:assertj-core:3.25.3'
+    testImplementation 'org.testng:testng:7.10.2'
+    testImplementation 'org.assertj:assertj-core:3.26.0'
 }
 
 test {

gradle.properties

@@ -1,3 +1,3 @@
-release_version=5.3.0
+release_version=5.4.0
 description='Cradle API'
 vcs_url=https://github.com/th2-net/cradleapi