Removed ComputePoolService, added feature flags and fixed pool time metric name

RSam25 · RSam25 · commit ec11efc42535 · 2026-01-27T15:19:19.000+11:00
diff --git a/conf/default-config.json b/conf/default-config.json
@@ -40,5 +40,7 @@
   "sharing_token_expiry_seconds": 2592000,
   "operator_type": "public",
   "enable_remote_config": true,
-  "uid_instance_id_prefix": "local-operator"
+  "uid_instance_id_prefix": "local-operator",
+  "enable_async_batch_request": true,
+  "compute_pool_thread_count": 12
 }
diff --git a/src/main/java/com/uid2/operator/Const.java b/src/main/java/com/uid2/operator/Const.java
@@ -39,5 +39,8 @@ public class Config extends com.uid2.shared.Const.Config {
         public static final String RuntimeConfigMetadataPathProp = "runtime_config_metadata_path";
 
         public static final String IdentityEnvironmentProp = "identity_environment";
+
+        public static final String ComputePoolThreadCountProp = "compute_pool_thread_count";
+        public static final String EnableAsyncBatchRequestProp = "enable_async_batch_request";
     }
 }
diff --git a/src/main/java/com/uid2/operator/Main.java b/src/main/java/com/uid2/operator/Main.java
@@ -342,10 +342,12 @@ private void run() throws Exception {
         this.createVertxEventLoopsMetric();
 
         // Create shared compute pool for CPU-intensive operations
-        final ComputePoolService computePoolService = new ComputePoolService(vertx);
+        final int computePoolSize = config.getInteger(Const.Config.ComputePoolThreadCountProp, Math.max(1, Runtime.getRuntime().availableProcessors() - 2));
+        final WorkerExecutor computeWorkerPool = vertx.createSharedWorkerExecutor("compute", computePoolSize);
+        LOGGER.info("Created compute worker pool with size: {}", computePoolSize);
 
         Supplier<Verticle> operatorVerticleSupplier = () -> {
-            UIDOperatorVerticle verticle = new UIDOperatorVerticle(configStore, config, this.clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, getKeyManager(), saltProvider, optOutStore, Clock.systemUTC(), _statsCollectorQueue, new SecureLinkValidatorService(this.serviceLinkProvider, this.serviceProvider), this.shutdownHandler::handleSaltRetrievalResponse, this.uidInstanceIdProvider, computePoolService);
+            UIDOperatorVerticle verticle = new UIDOperatorVerticle(configStore, config, this.clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, getKeyManager(), saltProvider, optOutStore, Clock.systemUTC(), _statsCollectorQueue, new SecureLinkValidatorService(this.serviceLinkProvider, this.serviceProvider), this.shutdownHandler::handleSaltRetrievalResponse, this.uidInstanceIdProvider, computeWorkerPool);
             return verticle;
         };
 
@@ -374,7 +376,9 @@ private void run() throws Exception {
                 })
                 .onFailure(t -> {
                     LOGGER.error("Failed to bootstrap operator: " + t.getMessage(), new Exception(t));
-                    computePoolService.close();
+                    if (computeWorkerPool != null) {
+                        computeWorkerPool.close();
+                    }
                     vertx.close();
                     System.exit(1);
                 });
@@ -504,7 +508,7 @@ private static Vertx createVertx() {
         VertxOptions vertxOptions = new VertxOptions()
             .setMetricsOptions(metricOptions)
             .setBlockedThreadCheckInterval(threadBlockedCheckInterval)
-            .setWorkerPoolSize(8);
+            .setWorkerPoolSize(12);
 
         return Vertx.vertx(vertxOptions);
     }
@@ -529,7 +533,7 @@ private static void setupMetrics(MicrometerMetricsOptions metricOptions) {
                     Objects.equals(id.getTag(Label.HTTP_CODE.toString()), "404")))
                 .meterFilter(new MeterFilter() {
                     private final String httpServerResponseTime = MetricsDomain.HTTP_SERVER.getPrefix() + MetricsNaming.v4Names().getHttpResponseTime();
-                    private final String poolQueueTime = MetricsDomain.HTTP_SERVER.getPrefix() + MetricsNaming.v4Names().getPoolQueueTime();
+                    private final String poolQueueTime = MetricsDomain.NAMED_POOLS.getPrefix() + MetricsNaming.v4Names().getPoolQueueTime();
 
                     @Override
                     public DistributionStatisticConfig configure(Meter.Id id, DistributionStatisticConfig config) {
diff --git a/src/main/java/com/uid2/operator/service/ComputePoolService.java b/src/main/java/com/uid2/operator/service/ComputePoolService.java
diff --git a/src/main/java/com/uid2/operator/vertx/UIDOperatorVerticle.java b/src/main/java/com/uid2/operator/vertx/UIDOperatorVerticle.java
@@ -43,6 +43,7 @@
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
+import io.vertx.core.WorkerExecutor;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.http.HttpServerResponse;
@@ -133,14 +134,16 @@ public class UIDOperatorVerticle extends AbstractVerticle {
     public static final long OPT_OUT_CHECK_CUTOFF_DATE = Instant.parse("2023-09-01T00:00:00.00Z").getEpochSecond();
     private final Handler<Boolean> saltRetrievalResponseHandler;
     private final int allowClockSkewSeconds;
-    private final ComputePoolService computePoolService;
+    private final WorkerExecutor computeWorkerPool;
     protected Map<Integer, Set<String>> siteIdToInvalidOriginsAndAppNames = new HashMap<>();
     protected boolean keySharingEndpointProvideAppNames;
     protected Instant lastInvalidOriginProcessTime = Instant.now();
 
     private final int optOutStatusMaxRequestSize;
     private final boolean optOutStatusApiEnabled;
 
+    private final boolean isAsyncBatchRequestsEnabled;
+
     //"Android" is from https://github.com/IABTechLab/uid2-android-sdk/blob/ff93ebf597f5de7d440a84f7015a334ba4138ede/sdk/src/main/java/com/uid2/UID2Client.kt#L46
     //"ios"/"tvos" is from https://github.com/IABTechLab/uid2-ios-sdk/blob/91c290d29a7093cfc209eca493d1fee80c17e16a/Sources/UID2/UID2Client.swift#L36-L38
     private static final List<String> SUPPORTED_IN_APP = Arrays.asList("Android", "ios", "tvos");
@@ -166,7 +169,7 @@ public UIDOperatorVerticle(IConfigStore configStore,
                                SecureLinkValidatorService secureLinkValidatorService,
                                Handler<Boolean> saltRetrievalResponseHandler,
                                UidInstanceIdProvider uidInstanceIdProvider,
-                               ComputePoolService computePoolService) {
+                               WorkerExecutor computeWorkerPool) {
         this.keyManager = keyManager;
         this.secureLinkValidatorService = secureLinkValidatorService;
         try {
@@ -200,7 +203,8 @@ public UIDOperatorVerticle(IConfigStore configStore,
         this.identityV3Enabled = config.getBoolean(IdentityV3Prop, false);
         this.disableOptoutToken = config.getBoolean(DisableOptoutTokenProp, false);
         this.uidInstanceIdProvider = uidInstanceIdProvider;
-        this.computePoolService = computePoolService;
+        this.computeWorkerPool = computeWorkerPool;
+        this.isAsyncBatchRequestsEnabled = config.getBoolean(EnableAsyncBatchRequestProp, false);
     }
 
     @Override
@@ -285,10 +289,6 @@ private void setUpEncryptedRoutes(Router mainRouter, BodyHandler bodyHandler) {
                 rc -> encryptedPayloadHandler.handleTokenRefresh(rc, this::handleTokenRefreshV2)));
         mainRouter.post(V2_TOKEN_VALIDATE.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> encryptedPayloadHandler.handle(rc, this::handleTokenValidateV2), Role.GENERATOR));
-        mainRouter.post(V2_IDENTITY_BUCKETS.toString()).handler(bodyHandler).handler(auth.handleV1(
-                rc -> encryptedPayloadHandler.handleAsync(rc, this::handleBucketsV2Async), Role.MAPPER));
-        mainRouter.post(V2_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
-                rc -> encryptedPayloadHandler.handleAsync(rc, this::handleIdentityMapV2Async), Role.MAPPER));
         mainRouter.post(V2_KEY_LATEST.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> encryptedPayloadHandler.handle(rc, this::handleKeysRequestV2), Role.ID_READER));
         mainRouter.post(V2_KEY_SHARING.toString()).handler(bodyHandler).handler(auth.handleV1(
@@ -306,8 +306,21 @@ private void setUpEncryptedRoutes(Router mainRouter, BodyHandler bodyHandler) {
         if (this.clientSideTokenGenerate)
             mainRouter.post(V2_TOKEN_CLIENTGENERATE.toString()).handler(bodyHandler).handler(this::handleClientSideTokenGenerate);
 
-        mainRouter.post(V3_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
-                rc -> encryptedPayloadHandler.handleAsync(rc, this::handleIdentityMapV3Async), Role.MAPPER));
+        if (isAsyncBatchRequestsEnabled) {
+            mainRouter.post(V2_IDENTITY_BUCKETS.toString()).handler(bodyHandler).handler(auth.handleV1(
+                    rc -> encryptedPayloadHandler.handleAsync(rc, this::handleBucketsV2Async), Role.MAPPER));
+            mainRouter.post(V2_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
+                    rc -> encryptedPayloadHandler.handleAsync(rc, this::handleIdentityMapV2Async), Role.MAPPER));
+            mainRouter.post(V3_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
+                    rc -> encryptedPayloadHandler.handleAsync(rc, this::handleIdentityMapV3Async), Role.MAPPER));
+        } else {
+            mainRouter.post(V2_IDENTITY_BUCKETS.toString()).handler(bodyHandler).handler(auth.handleV1(
+                    rc -> encryptedPayloadHandler.handle(rc, this::handleBucketsV2), Role.MAPPER));
+            mainRouter.post(V2_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
+                    rc -> encryptedPayloadHandler.handle(rc, this::handleIdentityMapV2), Role.MAPPER));
+            mainRouter.post(V3_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
+                    rc -> encryptedPayloadHandler.handle(rc, this::handleIdentityMapV3), Role.MAPPER));
+        }
     }
 
     private void handleClientSideTokenGenerate(RoutingContext rc) {
@@ -1041,8 +1054,9 @@ private Future handleLogoutAsyncV2(RoutingContext rc) {
     }
 
     private Future<Void> handleBucketsV2Async(RoutingContext rc) {
-        return computePoolService.executeBlocking(() -> {
+        return computeWorkerPool.executeBlocking(() -> {
             handleBucketsV2(rc);
+            return null;
         });
     }
 
@@ -1232,8 +1246,9 @@ private boolean validateServiceLink(RoutingContext rc) {
     }
 
     private Future<Void> handleIdentityMapV2Async(RoutingContext rc) {
-        return computePoolService.executeBlocking(() -> {
+        return computeWorkerPool.executeBlocking(() -> {
             handleIdentityMapV2(rc);
+            return null;
         });
     }
 
@@ -1301,8 +1316,9 @@ private InputUtil.InputVal[] getIdentityMapV2Input(RoutingContext rc) {
     }
 
     private Future<Void> handleIdentityMapV3Async(RoutingContext rc) {
-        return computePoolService.executeBlocking(() -> {
+        return computeWorkerPool.executeBlocking(() -> {
             handleIdentityMapV3(rc);
+            return null;
         });
     }
 
diff --git a/src/test/java/com/uid2/operator/ExtendedUIDOperatorVerticle.java b/src/test/java/com/uid2/operator/ExtendedUIDOperatorVerticle.java
@@ -2,7 +2,6 @@
 
 import com.uid2.operator.model.KeyManager;
 import com.uid2.operator.monitoring.IStatsCollectorQueue;
-import com.uid2.operator.service.ComputePoolService;
 import com.uid2.operator.service.IUIDOperatorService;
 import com.uid2.operator.service.SecureLinkValidatorService;
 import com.uid2.operator.store.IConfigStore;
@@ -12,6 +11,7 @@
 import com.uid2.shared.store.*;
 import com.uid2.shared.store.salt.ISaltProvider;
 import io.vertx.core.Handler;
+import io.vertx.core.WorkerExecutor;
 import io.vertx.core.json.JsonObject;
 
 import java.time.Clock;
@@ -35,8 +35,8 @@ public ExtendedUIDOperatorVerticle(IConfigStore configStore,
                                        SecureLinkValidatorService secureLinkValidationService,
                                        Handler<Boolean> saltRetrievalResponseHandler,
                                        UidInstanceIdProvider uidInstanceIdProvider,
-                                       ComputePoolService computePoolService) {
-        super(configStore, config, clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, keyManager, saltProvider, optOutStore, clock, statsCollectorQueue, secureLinkValidationService, saltRetrievalResponseHandler, uidInstanceIdProvider, computePoolService);
+                                       WorkerExecutor computeWorkerPool) {
+        super(configStore, config, clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, keyManager, saltProvider, optOutStore, clock, statsCollectorQueue, secureLinkValidationService, saltRetrievalResponseHandler, uidInstanceIdProvider, computeWorkerPool);
     }
 
     public IUIDOperatorService getIdService() {
diff --git a/src/test/java/com/uid2/operator/UIDOperatorVerticleTest.java b/src/test/java/com/uid2/operator/UIDOperatorVerticleTest.java
@@ -38,6 +38,7 @@
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
+import io.vertx.core.WorkerExecutor;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpHeaders;
 import io.vertx.core.json.JsonArray;
@@ -142,7 +143,7 @@ public class UIDOperatorVerticleTest {
     private ExtendedUIDOperatorVerticle uidOperatorVerticle;
     private RuntimeConfig runtimeConfig;
     private EncryptedTokenEncoder encoder;
-    private ComputePoolService computePoolService;
+    private WorkerExecutor computeWorkerPool;
 
     @BeforeEach
     void deployVerticle(Vertx vertx, VertxTestContext testContext, TestInfo testInfo) {
@@ -166,8 +167,8 @@ void deployVerticle(Vertx vertx, VertxTestContext testContext, TestInfo testInfo
 
         this.uidInstanceIdProvider = new UidInstanceIdProvider("test-instance", "id");
 
-        this.computePoolService = new ComputePoolService(vertx);
-        this.uidOperatorVerticle = new ExtendedUIDOperatorVerticle(configStore, config, config.getBoolean("client_side_token_generate"), siteProvider, clientKeyProvider, clientSideKeypairProvider, new KeyManager(keysetKeyStore, keysetProvider), saltProvider, optOutStore, clock, statsCollectorQueue, secureLinkValidatorService, shutdownHandler::handleSaltRetrievalResponse, uidInstanceIdProvider, this.computePoolService);
+        this.computeWorkerPool = vertx.createSharedWorkerExecutor("compute", 4);
+        this.uidOperatorVerticle = new ExtendedUIDOperatorVerticle(configStore, config, config.getBoolean("client_side_token_generate"), siteProvider, clientKeyProvider, clientSideKeypairProvider, new KeyManager(keysetKeyStore, keysetProvider), saltProvider, optOutStore, clock, statsCollectorQueue, secureLinkValidatorService, shutdownHandler::handleSaltRetrievalResponse, uidInstanceIdProvider, this.computeWorkerPool);
         vertx.deployVerticle(uidOperatorVerticle, testContext.succeeding(id -> testContext.completeNow()));
 
         this.registry = new SimpleMeterRegistry();
@@ -179,8 +180,8 @@ void deployVerticle(Vertx vertx, VertxTestContext testContext, TestInfo testInfo
     @AfterEach
     void teardown() {
         Metrics.globalRegistry.remove(registry);
-        if (computePoolService != null) {
-            computePoolService.close();
+        if (computeWorkerPool != null) {
+            computeWorkerPool.close();
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -40,5 +40,7 @@`
`40`	`40`	`"sharing_token_expiry_seconds": 2592000,`
`41`	`41`	`"operator_type": "public",`
`42`	`42`	`"enable_remote_config": true,`
`43`		`- "uid_instance_id_prefix": "local-operator"`
	`43`	`+ "uid_instance_id_prefix": "local-operator",`
	`44`	`+ "enable_async_batch_request": true,`
	`45`	`+ "compute_pool_thread_count": 12`
`44`	`46`	`}`
Original file line number	Diff line number	Diff line change
`@@ -39,5 +39,8 @@ public class Config extends com.uid2.shared.Const.Config {`
`39`	`39`	`public static final String RuntimeConfigMetadataPathProp = "runtime_config_metadata_path";`
`40`	`40`
`41`	`41`	`public static final String IdentityEnvironmentProp = "identity_environment";`
	`42`	`+`
	`43`	`+ public static final String ComputePoolThreadCountProp = "compute_pool_thread_count";`
	`44`	`+ public static final String EnableAsyncBatchRequestProp = "enable_async_batch_request";`
`42`	`45`	`}`
`43`	`46`	`}`