From 1fc03f6d8ed55aebb5175c45e34323873287bb51 Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Wed, 13 May 2026 19:51:54 +0530 Subject: [PATCH 1/9] US1124088: job-service Payload Batching --- .../JobServicePayloadBatchingEndToEndIT.java | 450 +++++++++++++++ .../JobServicePayloadBatchingIT.java | 544 ++++++++++++++++++ job-service-scheduled-executor/pom.xml | 6 + .../scheduledexecutor/QueueServices.java | 148 ++++- .../batching/PayloadBatchingService.java | 194 +++++++ .../batching/SubdocumentBatchSplitter.java | 152 +++++ .../batching/SubtaskIdGenerator.java | 96 ++++ .../batching/PayloadBatchingTest.java | 523 +++++++++++++++++ 8 files changed, 2110 insertions(+), 3 deletions(-) create mode 100644 job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java create mode 100644 job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java create mode 100644 job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java create mode 100644 job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java create mode 100644 job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubtaskIdGenerator.java create mode 100644 job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java diff --git a/job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java b/job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java new file mode 100644 index 000000000..c1441e38a --- /dev/null +++ b/job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java @@ -0,0 +1,450 @@ +/* + * Copyright 2016-2026 Open Text. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.jobservice.acceptancetests; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.cafapi.common.api.BootstrapConfiguration; +import com.github.cafapi.common.api.ConfigurationSource; +import com.github.cafapi.common.bootstrapconfigs.system.SystemBootstrapConfiguration; +import com.github.cafapi.common.util.naming.ServicePath; +import com.github.jobservice.client.api.JobsApi; +import com.github.jobservice.client.model.Job; +import com.github.jobservice.client.model.JobStatus; +import com.github.jobservice.client.model.NewJob; +import com.github.jobservice.client.model.WorkerAction; +import com.github.jobservice.job.client.ApiClient; +import com.github.jobservice.job.client.ApiException; +import com.github.workerframework.api.TaskMessage; +import com.github.workerframework.queues.rabbit.RabbitWorkerQueueConfiguration; +import com.github.workerframework.testing.ExecutionContext; +import com.github.workerframework.testing.SettingNames; +import com.github.workerframework.testing.util.QueueManager; +import com.github.workerframework.testing.util.QueueServices; +import com.github.workerframework.testing.util.QueueServicesFactory; +import com.github.workerframework.testing.util.SettingsProvider; +import com.github.workerframework.testing.util.WorkerServices; +import com.github.workerframework.util.rabbitmq.RabbitUtil; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import java.io.IOException; +import java.net.URISyntaxException; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Timer; +import java.util.TimerTask; +import java.util.UUID; +import java.util.concurrent.TimeoutException; +import java.util.function.Supplier; + +import static org.testng.Assert.*; + +/** + * End-to-End integration tests for Payload Batching feature. + * Tests complete job lifecycle with batched payloads. + */ +public class JobServicePayloadBatchingEndToEndIT +{ + private static final Logger LOG = LoggerFactory.getLogger(JobServicePayloadBatchingEndToEndIT.class); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final int BATCH_SIZE = 200; + private static final String JOB_CORRELATION_ID = "1"; + private static final long JOB_STATUS_CHECK_TIMEOUT_MS = 60000; + private static final long JOB_STATUS_CHECK_SLEEP_MS = 500; + private static final long DEFAULT_TIMEOUT_MS = 120000; + private static final String SUBDOCUMENT_BATCHER_PREFIX = "DocumentWorkerSubdocumentBatcher() "; + private static final String RABBIT_PROP_QUEUE_TYPE = "x-queue-type"; + private static final String RABBIT_PROP_QUEUE_TYPE_QUORUM = "quorum"; + + private static ServicePath servicePath; + private static WorkerServices workerServices; + private static ConfigurationSource configurationSource; + private static RabbitWorkerQueueConfiguration rabbitConfiguration; + private static JobsApi jobsApi; + private static Connection rabbitConn; + + private String defaultPartitionId; + private QueueManager testQueueManager; + + @BeforeClass + public static void setupClass() throws Exception + { + final BootstrapConfiguration bootstrap = new SystemBootstrapConfiguration(); + servicePath = bootstrap.getServicePath(); + workerServices = WorkerServices.getDefault(); + configurationSource = workerServices.getConfigurationSource(); + rabbitConfiguration = configurationSource.getConfiguration(RabbitWorkerQueueConfiguration.class); + rabbitConfiguration.getRabbitConfiguration().setRabbitHost( + SettingsProvider.defaultProvider.getSetting(SettingNames.dockerHostAddress)); + rabbitConfiguration.getRabbitConfiguration().setRabbitPort( + Integer.parseInt(SettingsProvider.defaultProvider.getSetting(SettingNames.rabbitmqNodePort))); + + // Create RabbitMQ connection for queue operations + rabbitConn = RabbitUtil.createRabbitConnection(rabbitConfiguration.getRabbitConfiguration()); + + final String connectionString = System.getenv("webserviceurl"); + final ApiClient client = new ApiClient(); + client.setBasePath(connectionString); + final SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ"); + client.setDateFormat(f); + jobsApi = new JobsApi(client); + } + + @BeforeMethod + public void setupMethod() + { + defaultPartitionId = UUID.randomUUID().toString(); + } + + @AfterMethod + public void tearDownMethod() throws IOException + { + if (testQueueManager != null) { + testQueueManager.close(); + testQueueManager = null; + } + } + + // Batched job completion tracking - verifies message count and job status + @Test + public void testBatchedJob_CompletionTracking() throws Exception + { + final String queueName = "payload-batching-e2e-completion-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE * 3; // 3 batches = 3 messages + final int expectedBatchCount = 3; + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName); + + LOG.info("Creating batched job {} with {} subdocuments ({} batches)", + jobId, subdocCount, expectedBatchCount); + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, expectedBatchCount); + + // Create the job + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, JOB_CORRELATION_ID); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + + // Verify correct number of batched messages received + assertEquals(received.size(), expectedBatchCount, + "Should receive " + expectedBatchCount + " batched messages on the queue"); + LOG.info("Received {} messages on queue {} as expected", received.size(), queueName); + + // Verify job was created and is processing + final Job job = jobsApi.getJob(defaultPartitionId, jobId, JOB_CORRELATION_ID); + assertNotNull(job, "Job should exist"); + assertEquals(job.getId(), jobId); + assertNotNull(job.getStatus(), "Job should have a status"); + + // Job should be in Waiting or Active state (messages are queued, waiting for worker) + assertTrue(job.getStatus() == JobStatus.WAITING || job.getStatus() == JobStatus.ACTIVE, + "Job should be in Waiting or Active status, but was: " + job.getStatus()); + } + + // Batched job progress reporting - verifies messages published and tracks progress + @Test + public void testBatchedJob_ProgressReporting() throws Exception + { + final String queueName = "payload-batching-e2e-progress-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE * 3; // 3 batches = 3 messages + final int expectedBatchCount = 3; + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName); + + LOG.info("Creating job {} for progress tracking test with {} batches", jobId, expectedBatchCount); + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, expectedBatchCount); + + // Create the job + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, JOB_CORRELATION_ID); + + // Verify all batched messages are published + final List received = messageSupplier.get(); + assertEquals(received.size(), expectedBatchCount, + "Should receive " + expectedBatchCount + " batched messages"); + LOG.info("All {} messages published to queue {}", received.size(), queueName); + + // Verify job status + Job job = jobsApi.getJob(defaultPartitionId, jobId, JOB_CORRELATION_ID); + assertNotNull(job.getStatus(), "Job should have status"); + LOG.info("Job {} status: {}, percentageComplete: {}", + jobId, job.getStatus(), job.getPercentageComplete()); + + // Progress will remain at 0 or initial state until workers complete subtasks + assertTrue(job.getStatus() == JobStatus.WAITING || job.getStatus() == JobStatus.ACTIVE, + "Job should be Waiting or Active after messages published"); + } + + // Batched job cancellation - verifies job can be cancelled and checks status + @Test + public void testBatchedJob_Cancellation() throws Exception + { + final String queueName = "payload-batching-e2e-cancel-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE * 5; // 5 batches - larger to allow time for cancellation + final int expectedBatchCount = 5; + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName); + + LOG.info("Creating job {} for cancellation test with {} batches", jobId, expectedBatchCount); + + // Start listening for messages + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, expectedBatchCount); + + // Create the job + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, JOB_CORRELATION_ID); + + // Wait for all messages to be published + final List received = messageSupplier.get(); + assertEquals(received.size(), expectedBatchCount, + "Should receive " + expectedBatchCount + " batched messages before cancellation"); + LOG.info("Received {} messages, now cancelling job {}", received.size(), jobId); + + // Get job state before cancellation + Job jobBeforeCancel = jobsApi.getJob(defaultPartitionId, jobId, JOB_CORRELATION_ID); + LOG.info("Job {} status before cancel: {}, percentageComplete: {}", + jobId, jobBeforeCancel.getStatus(), jobBeforeCancel.getPercentageComplete()); + + // Cancel the job + jobsApi.cancelJob(defaultPartitionId, jobId, JOB_CORRELATION_ID); + + // Verify cancellation + final Job job = jobsApi.getJob(defaultPartitionId, jobId, JOB_CORRELATION_ID); + assertEquals(job.getStatus(), JobStatus.CANCELLED, "Job should be cancelled"); + + // Log the percentage complete at cancellation time + LOG.info("Job {} successfully cancelled. Final percentageComplete: {}", + jobId, job.getPercentageComplete()); + } + + // Small subdocuments - no batching - verifies single message and job status + @Test + public void testSmallSubdocuments_NoBatching() throws Exception + { + final String queueName = "payload-batching-e2e-small-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE / 2; // Below threshold - no batching + final int expectedMessageCount = 1; // Single message expected + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName); + + LOG.info("Creating non-batched job {} with {} subdocuments (below threshold)", jobId, subdocCount); + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, expectedMessageCount); + + // Create the job + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, JOB_CORRELATION_ID); + + // Wait for message + final List received = messageSupplier.get(); + + // Verify single message received (no batching) + assertEquals(received.size(), expectedMessageCount, + "Should receive exactly 1 message for small subdocument count"); + + // Verify job created and status + final Job job = jobsApi.getJob(defaultPartitionId, jobId, JOB_CORRELATION_ID); + assertNotNull(job, "Job should exist"); + assertEquals(job.getId(), jobId); + assertTrue(job.getStatus() == JobStatus.WAITING || job.getStatus() == JobStatus.ACTIVE, + "Job should be in Waiting or Active status"); + + LOG.info("Non-batched job {} created with 1 message, status: {}", jobId, job.getStatus()); + } + + // === HELPER METHODS === + + private NewJob createDocumentWorkerJob(final String jobId, final int subdocCount, final String queueName) throws Exception + { + // Declare the queue first + declareQueue(queueName); + + // taskData structure matches actual Job Definition format + final Map taskData = new HashMap<>(); + + // document with fields and subdocuments array + final Map document = new HashMap<>(); + // Add fields inside document (like targetId, destinationId in real payloads) + final Map documentFields = new HashMap<>(); + documentFields.put("targetId", List.of(Map.of("data", "2"))); + documentFields.put("destinationId", List.of(Map.of("data", "4"))); + document.put("fields", documentFields); + + final List subdocs = new ArrayList<>(); + for (int i = 1; i <= subdocCount; i++) { + // Each subdoc has fields structure like actual format + final Map subdoc = new HashMap<>(); + final Map fields = new HashMap<>(); + fields.put("_id", List.of(Map.of("data", String.valueOf(i)))); + fields.put("_index", List.of(Map.of("data", "test-index"))); + fields.put("_routing", List.of(Map.of("data", String.valueOf(i)))); + subdoc.put("fields", fields); + subdocs.add(subdoc); + } + document.put("subdocuments", subdocs); + taskData.put("document", document); + + // customData at taskData root level + final Map customData = new HashMap<>(); + customData.put("tenantId", "testTenant"); + customData.put("workflowName", "test-workflow"); + customData.put("timestamp", String.valueOf(System.currentTimeMillis())); + taskData.put("customData", customData); + + // scripts at taskData root level + taskData.put("scripts", new ArrayList<>()); + + final WorkerAction task = new WorkerAction(); + task.setTaskClassifier("DocumentWorkerTask"); + task.setTaskApiVersion(2); + task.setTaskData(OBJECT_MAPPER.writeValueAsString(taskData)); + task.setTaskDataEncoding(WorkerAction.TaskDataEncodingEnum.UTF8); + // Add batching prefix for opt-in + task.setTaskPipe(SUBDOCUMENT_BATCHER_PREFIX + queueName); + task.setTargetPipe(queueName + "-out"); + + final NewJob newJob = new NewJob(); + newJob.setName("PayloadBatchingE2E_" + jobId); + newJob.setDescription("End-to-end payload batching test"); + newJob.setTask(task); + + return newJob; + } + + private void declareQueue(final String queueName) throws Exception + { + final Channel channel = rabbitConn.createChannel(); + final Map args = new HashMap<>(); + args.put(RABBIT_PROP_QUEUE_TYPE, RABBIT_PROP_QUEUE_TYPE_QUORUM); + channel.queueDeclare(queueName, true, false, false, args); + channel.close(); + } + + private QueueManager getQueueManager(final String queueName) + throws IOException, TimeoutException, URISyntaxException, NoSuchAlgorithmException, KeyManagementException + { + final QueueServices queueServices = QueueServicesFactory.create( + rabbitConfiguration, queueName, workerServices.getCodec()); + final boolean debugEnabled = SettingsProvider.defaultProvider.getBooleanSetting( + SettingNames.createDebugMessage, false); + return new QueueManager(queueServices, workerServices, debugEnabled); + } + + /** + * Starts collecting messages from the queue and returns a Supplier that blocks until + * the expected number of messages are received or timeout occurs. + */ + private Supplier> startCollectingMessages(final QueueManager queueManager, final int expectedCount) + throws Exception + { + final ExecutionContext context = new ExecutionContext(false); + context.initializeContext(); + + final List result = new ArrayList<>(); + final int[] receivedCount = {0}; + + // Set up timeout timer + final Timer timer = getTimer(context); + + // Start consuming messages + queueManager.start(message -> { + result.add(message); + receivedCount[0]++; + LOG.debug("Received message {} of {}", receivedCount[0], expectedCount); + if (receivedCount[0] >= expectedCount) { + timer.cancel(); + context.finishedSuccessfully(); + } + }); + + // Return supplier that blocks on test result + return () -> { + assertTrue(context.getTestResult().isSuccess(), + "Should receive " + expectedCount + " messages from queue"); + return result; + }; + } + + private Timer getTimer(final ExecutionContext context) + { + final String timeoutSetting = SettingsProvider.defaultProvider.getSetting(SettingNames.timeOutMs); + final long timeout = timeoutSetting == null ? DEFAULT_TIMEOUT_MS : Long.parseLong(timeoutSetting); + final Timer timer = new Timer(); + timer.schedule(new TimerTask() { + @Override + public void run() { + context.testRunsTimedOut(); + } + }, timeout); + return timer; + } + + private void waitForJobStatus(final String jobId, final JobStatus expectedStatus, final long timeoutMs) + throws ApiException, InterruptedException + { + final long startTime = System.currentTimeMillis(); + Job job; + + while (System.currentTimeMillis() - startTime < timeoutMs) { + job = jobsApi.getJob(defaultPartitionId, jobId, JOB_CORRELATION_ID); + + if (job.getStatus() == expectedStatus) { + LOG.info("Job {} reached status {}", jobId, expectedStatus); + return; + } + + // Also check for terminal states + if (expectedStatus != JobStatus.COMPLETED && + expectedStatus != JobStatus.CANCELLED && + expectedStatus != JobStatus.FAILED) { + if (job.getStatus() == JobStatus.COMPLETED || + job.getStatus() == JobStatus.CANCELLED || + job.getStatus() == JobStatus.FAILED) { + LOG.info("Job {} reached terminal status {}", jobId, job.getStatus()); + return; + } + } + + Thread.sleep(JOB_STATUS_CHECK_SLEEP_MS); + } + + LOG.warn("Job {} did not reach status {} within timeout", jobId, expectedStatus); + } +} + + + diff --git a/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java b/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java new file mode 100644 index 000000000..28bdaaa7e --- /dev/null +++ b/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java @@ -0,0 +1,544 @@ +/* + * Copyright 2016-2026 Open Text. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.jobservice.containertests; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.cafapi.common.api.BootstrapConfiguration; +import com.github.cafapi.common.bootstrapconfigs.system.SystemBootstrapConfiguration; +import com.github.cafapi.common.util.naming.ServicePath; +import com.github.jobservice.client.api.JobsApi; +import com.github.jobservice.client.model.Job; +import com.github.jobservice.client.model.NewJob; +import com.github.jobservice.client.model.WorkerAction; +import com.github.jobservice.job.client.ApiClient; +import com.github.workerframework.api.TaskMessage; +import com.github.workerframework.api.TrackingInfo; +import com.github.workerframework.queues.rabbit.RabbitWorkerQueueConfiguration; +import com.github.workerframework.testing.ExecutionContext; +import com.github.workerframework.testing.SettingNames; +import com.github.workerframework.testing.util.QueueManager; +import com.github.workerframework.testing.util.QueueServices; +import com.github.workerframework.testing.util.QueueServicesFactory; +import com.github.workerframework.testing.util.SettingsProvider; +import com.github.workerframework.testing.util.WorkerServices; +import com.github.workerframework.util.rabbitmq.RabbitUtil; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.annotations.AfterTest; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.BeforeTest; +import org.testng.annotations.Test; + +import java.io.IOException; +import java.net.URISyntaxException; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Timer; +import java.util.TimerTask; +import java.util.UUID; +import java.util.concurrent.TimeoutException; +import java.util.function.Supplier; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +import static org.testng.Assert.*; + +/** + * Integration tests for Payload Batching feature. + * Tests message delivery with containers for batched DocumentWorkerTask payloads. + */ +public class JobServicePayloadBatchingIT +{ + private static final Logger LOG = LoggerFactory.getLogger(JobServicePayloadBatchingIT.class); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final int BATCH_SIZE = 200; + private static final long DEFAULT_TIMEOUT_MS = 120000; + private static final Pattern SUBTASK_ID_PATTERN = Pattern.compile(".*\\.\\d+\\*?$"); + private static final String SUBDOCUMENT_BATCHER_PREFIX = "DocumentWorkerSubdocumentBatcher() "; + + private String connectionString; + private String defaultPartitionId; + private ApiClient client = new ApiClient(); + private JobsApi jobsApi; + private QueueManager testQueueManager; + private Connection rabbitConn; + + private static ServicePath servicePath; + private static WorkerServices workerServices; + private static RabbitWorkerQueueConfiguration rabbitConfiguration; + + @BeforeTest + public void setup() throws Exception + { + connectionString = System.getenv("webserviceurl"); + client.setBasePath(connectionString); + final SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ"); + client.setDateFormat(f); + jobsApi = new JobsApi(client); + + final BootstrapConfiguration bootstrap = new SystemBootstrapConfiguration(); + servicePath = bootstrap.getServicePath(); + workerServices = WorkerServices.getDefault(); + rabbitConfiguration = workerServices.getConfigurationSource() + .getConfiguration(RabbitWorkerQueueConfiguration.class); + rabbitConfiguration.getRabbitConfiguration().setRabbitHost( + SettingsProvider.defaultProvider.getSetting(SettingNames.dockerHostAddress)); + rabbitConfiguration.getRabbitConfiguration().setRabbitPort( + Integer.parseInt(SettingsProvider.defaultProvider.getSetting(SettingNames.rabbitmqNodePort))); + rabbitConn = RabbitUtil.createRabbitConnection(rabbitConfiguration.getRabbitConfiguration()); + } + + @AfterTest + public void tearDown() throws IOException + { + if (testQueueManager != null) { + testQueueManager.close(); + } + if (rabbitConn != null) { + rabbitConn.close(); + } + } + + @BeforeMethod + public void setupMethod() + { + defaultPartitionId = UUID.randomUUID().toString(); + } + + // Batched messages received on queue + @Test + public void testBatchedMessages_ReceivedOnQueue() throws Exception + { + final String queueName = "payload-batching-test-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE * 3 + 50; // 3.25 batches = 4 messages + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, true); + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 4); + + // Create the job (messages will be published) + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + assertEquals(received.size(), 4, "Should receive 4 batched messages"); + } + + // Batched messages have correct subtask IDs in BOTH fields + @Test + public void testBatchedMessages_CorrectSubtaskIds() throws Exception + { + final String queueName = "payload-batching-subtaskid-test-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE * 2 + 50; // 2.25 batches = 3 messages + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, true); + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 3); + + // Create the job (messages will be published) + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + assertEquals(received.size(), 3); + + // Verify subtask IDs in both fields + for (int i = 0; i < received.size(); i++) { + final TaskMessage msg = received.get(i); + final String taskId = msg.getTaskId(); + final TrackingInfo tracking = msg.getTracking(); + final String jobTaskId = tracking.getJobTaskId(); + + // Both should have subtask suffix + assertTrue(SUBTASK_ID_PATTERN.matcher(taskId).matches(), + "TaskMessage.taskId should have subtask suffix: " + taskId); + assertTrue(SUBTASK_ID_PATTERN.matcher(jobTaskId).matches(), + "TrackingInfo.jobTaskId should have subtask suffix: " + jobTaskId); + + // Extract and compare suffixes + final String taskSuffix = taskId.substring(taskId.lastIndexOf('.')); + final String jobTaskSuffix = jobTaskId.substring(jobTaskId.lastIndexOf('.')); + assertEquals(taskSuffix, jobTaskSuffix, "Suffixes should match"); + + // Last batch should have asterisk + if (i == received.size() - 1) { + assertTrue(taskId.endsWith("*"), "Last taskId should end with *"); + assertTrue(jobTaskId.endsWith("*"), "Last jobTaskId should end with *"); + } else { + assertFalse(taskId.endsWith("*"), "Non-final taskId should not end with *"); + assertFalse(jobTaskId.endsWith("*"), "Non-final jobTaskId should not end with *"); + } + } + } + + // Batched messages have valid payload structure + @Test + @SuppressWarnings("unchecked") + public void testBatchedMessages_PayloadStructure() throws Exception + { + final String queueName = "payload-batching-payload-test-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE + 50; // 2 messages + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, true); + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 2); + + // Create the job (messages will be published) + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + assertEquals(received.size(), 2); + + // Expected document fields (same as created in createDocumentWorkerJob) + final Map expectedDocumentFields = new HashMap<>(); + expectedDocumentFields.put("targetId", List.of(Map.of("data", "2"))); + expectedDocumentFields.put("destinationId", List.of(Map.of("data", "4"))); + + // Expected customData (same as created in createDocumentWorkerJob) + final String expectedTenantId = "testTenant"; + final String expectedWorkflowName = "test-workflow"; + + int totalSubdocs = 0; + for (final TaskMessage msg : received) { + // Parse taskData + final Map taskData = OBJECT_MAPPER.readValue(msg.getTaskData(), Map.class); + assertNotNull(taskData, "TaskData should not be null"); + assertTrue(taskData.containsKey("document"), "Should have document field"); + + final Map document = (Map) taskData.get("document"); + + // Verify fields inside document are preserved with correct values + assertTrue(document.containsKey("fields"), "Document should have fields"); + final Map documentFields = (Map) document.get("fields"); + assertNotNull(documentFields, "Document fields should not be null"); + assertEquals(documentFields, expectedDocumentFields, "Document fields should match original values"); + + // Verify customData is preserved with correct values + assertTrue(taskData.containsKey("customData"), "Should have customData field"); + final Map customData = (Map) taskData.get("customData"); + assertNotNull(customData, "customData should not be null"); + assertEquals(customData.get("tenantId"), expectedTenantId, "tenantId should match"); + assertEquals(customData.get("workflowName"), expectedWorkflowName, "workflowName should match"); + assertNotNull(customData.get("timestamp"), "timestamp should be present"); + + // Verify scripts is preserved + assertTrue(taskData.containsKey("scripts"), "Should have scripts field"); + final List scripts = (List) taskData.get("scripts"); + assertNotNull(scripts, "scripts should not be null"); + assertTrue(scripts.isEmpty(), "scripts should be empty list"); + + // Verify subdocuments + assertTrue(document.containsKey("subdocuments"), "Should have subdocuments"); + + final List subdocs = (List) document.get("subdocuments"); + assertNotNull(subdocs, "Subdocuments should not be null"); + assertTrue(subdocs.size() <= BATCH_SIZE, "Batch size should not exceed " + BATCH_SIZE); + + totalSubdocs += subdocs.size(); + } + + assertEquals(totalSubdocs, subdocCount, "Total subdocuments should match original"); + } + + // Non-DocumentWorkerTask job - single message + @Test + public void testNonBatchedJob_SingleMessage() throws Exception + { + final String queueName = "payload-batching-nonbatch-test-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createNonBatchedJob(jobId, queueName); + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 1); + + // Create the job (messages will be published) + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + assertEquals(received.size(), 1, "Should receive single message"); + + // Verify no subtask suffix + final String taskId = received.get(0).getTaskId(); + assertFalse(taskId.contains("."), "Non-batched taskId should not have subtask suffix"); + } + + // Small DocumentWorkerTask - single message + @Test + public void testSmallDocumentWorkerTask_SingleMessage() throws Exception + { + final String queueName = "payload-batching-small-test-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE / 2; // Below threshold + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, true); + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 1); + + // Create the job (messages will be published) + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + assertEquals(received.size(), 1, "Should receive single message for small subdocs"); + + // Verify no subtask suffix + final String taskId = received.get(0).getTaskId(); + assertFalse(taskId.contains(".") && SUBTASK_ID_PATTERN.matcher(taskId).matches(), + "Small DocumentWorkerTask should not have subtask suffix"); + } + + // DocumentWorkerTask WITHOUT prefix but with large subdocs - single message (not opted in) + @Test + public void testDocumentWorkerTask_NoPrefix_SingleMessage() throws Exception + { + final String queueName = "payload-batching-noprefix-test-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE * 3; // Above threshold but no prefix + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, false); // WITHOUT prefix + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 1); + + // Create the job (messages will be published) + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + assertEquals(received.size(), 1, "Should receive single message when prefix not present (not opted in)"); + + // Verify no subtask suffix + final String taskId = received.get(0).getTaskId(); + assertFalse(SUBTASK_ID_PATTERN.matcher(taskId).matches(), + "DocumentWorkerTask without prefix should not have subtask suffix"); + } + + // Verify task pipe in batch messages has prefix stripped + @Test + public void testBatchedMessages_TaskPipeStripped() throws Exception + { + final String queueName = "payload-batching-stripped-test-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE + 50; // 2 messages + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, true); // WITH prefix + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 2); + + // Create the job (messages will be published) + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + assertEquals(received.size(), 2); + + // Verify the task pipe in each message does NOT have the prefix + for (final TaskMessage msg : received) { + final String toQueue = msg.getTo(); + assertFalse(toQueue.startsWith(SUBDOCUMENT_BATCHER_PREFIX), + "Task pipe should have prefix stripped, got: " + toQueue); + assertEquals(toQueue, queueName, "Task pipe should be the original queue name"); + } + } + + // === HELPER METHODS === + + /** + * Creates a DocumentWorkerTask job with optional batching prefix. + * @param jobId Job ID + * @param subdocCount Number of subdocuments + * @param queueName Target queue name + * @param withBatchingPrefix If true, adds DocumentWorkerSubdocumentBatcher() prefix to task pipe + */ + private NewJob createDocumentWorkerJob(final String jobId, final int subdocCount, final String queueName, + final boolean withBatchingPrefix) throws Exception + { + declareQueue(queueName); + + // taskData structure matches actual Job Definition format + final Map taskData = new HashMap<>(); + + // document with fields and subdocuments array + final Map document = new HashMap<>(); + // Add fields inside document (like targetId, destinationId in real payloads) + final Map documentFields = new HashMap<>(); + documentFields.put("targetId", List.of(Map.of("data", "2"))); + documentFields.put("destinationId", List.of(Map.of("data", "4"))); + document.put("fields", documentFields); + + final List subdocs = new ArrayList<>(); + for (int i = 1; i <= subdocCount; i++) { + // Each subdoc has fields structure like actual format + final Map subdoc = new HashMap<>(); + final Map fields = new HashMap<>(); + fields.put("_id", List.of(Map.of("data", String.valueOf(i)))); + fields.put("_index", List.of(Map.of("data", "test-index"))); + fields.put("_routing", List.of(Map.of("data", String.valueOf(i)))); + subdoc.put("fields", fields); + subdocs.add(subdoc); + } + document.put("subdocuments", subdocs); + taskData.put("document", document); + + // customData at taskData root level + final Map customData = new HashMap<>(); + customData.put("tenantId", "testTenant"); + customData.put("workflowName", "test-workflow"); + customData.put("timestamp", String.valueOf(System.currentTimeMillis())); + taskData.put("customData", customData); + + // scripts at taskData root level + taskData.put("scripts", new ArrayList<>()); + + final WorkerAction task = new WorkerAction(); + task.setTaskClassifier("DocumentWorkerTask"); + task.setTaskApiVersion(2); + task.setTaskData(OBJECT_MAPPER.writeValueAsString(taskData)); + task.setTaskDataEncoding(WorkerAction.TaskDataEncodingEnum.UTF8); + // Add prefix if batching is opted-in + task.setTaskPipe(withBatchingPrefix ? SUBDOCUMENT_BATCHER_PREFIX + queueName : queueName); + task.setTargetPipe(queueName + "-out"); + + final NewJob newJob = new NewJob(); + newJob.setName("BatchingTest_" + jobId); + newJob.setDescription("Payload batching integration test"); + newJob.setTask(task); + + return newJob; + } + + private NewJob createNonBatchedJob(final String jobId, final String queueName) throws Exception + { + declareQueue(queueName); + + final WorkerAction task = new WorkerAction(); + task.setTaskClassifier("SomeOtherWorker"); + task.setTaskApiVersion(1); + task.setTaskData("{\"data\": \"test\"}"); + task.setTaskDataEncoding(WorkerAction.TaskDataEncodingEnum.UTF8); + task.setTaskPipe(queueName); + task.setTargetPipe(queueName + "-out"); + + final NewJob newJob = new NewJob(); + newJob.setName("NonBatchingTest_" + jobId); + newJob.setDescription("Non-batched job test"); + newJob.setTask(task); + + return newJob; + } + + private void declareQueue(final String queueName) throws Exception + { + final Channel channel = rabbitConn.createChannel(); + final Map args = new HashMap<>(); + args.put(JobServiceConnectionUtil.RABBIT_PROP_QUEUE_TYPE, + JobServiceConnectionUtil.RABBIT_PROP_QUEUE_TYPE_QUORUM); + channel.queueDeclare(queueName, true, false, false, args); + channel.close(); + } + + private QueueManager getQueueManager(final String queueName) + throws IOException, TimeoutException, URISyntaxException, NoSuchAlgorithmException, KeyManagementException + { + final QueueServices queueServices = QueueServicesFactory.create( + rabbitConfiguration, queueName, workerServices.getCodec()); + final boolean debugEnabled = SettingsProvider.defaultProvider.getBooleanSetting( + SettingNames.createDebugMessage, false); + return new QueueManager(queueServices, workerServices, debugEnabled); + } + + /** + * Starts collecting messages from the queue and returns a Supplier that blocks until + * the expected number of messages are received or timeout occurs. + * + * @param queueManager the queue manager to consume messages from + * @param expectedCount the number of messages to wait for + * @return Supplier that when called, blocks until messages are received and returns them + * @throws Exception if setup fails + */ + private Supplier> startCollectingMessages(final QueueManager queueManager, final int expectedCount) + throws Exception + { + final ExecutionContext context = new ExecutionContext(false); + context.initializeContext(); + + final List result = new ArrayList<>(); + final int[] receivedCount = {0}; + + // Set up timeout timer (uses settings or default) + final Timer timer = getTimer(context); + + // Start consuming messages + queueManager.start(message -> { + result.add(message); + receivedCount[0]++; + LOG.debug("Received message {} of {}", receivedCount[0], expectedCount); + if (receivedCount[0] >= expectedCount) { + timer.cancel(); + context.finishedSuccessfully(); + } + }); + + // Return supplier that blocks on test result + return () -> { + assertTrue(context.getTestResult().isSuccess(), + "Should receive " + expectedCount + " messages from queue"); + return result; + }; + } + + private Timer getTimer(final ExecutionContext context) + { + final String timeoutSetting = SettingsProvider.defaultProvider.getSetting(SettingNames.timeOutMs); + final long timeout = timeoutSetting == null ? DEFAULT_TIMEOUT_MS : Long.parseLong(timeoutSetting); + final Timer timer = new Timer(); + timer.schedule(new TimerTask() { + @Override + public void run() { + context.testRunsTimedOut(); + } + }, timeout); + return timer; + } +} + + + diff --git a/job-service-scheduled-executor/pom.xml b/job-service-scheduled-executor/pom.xml index 7c180bd9b..2e1e623a1 100644 --- a/job-service-scheduled-executor/pom.xml +++ b/job-service-scheduled-executor/pom.xml @@ -125,6 +125,12 @@ guava + + + org.junit.jupiter + junit-jupiter-api + test + diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java index 6aeff23c0..5eea0484a 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java @@ -34,10 +34,15 @@ import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.Date; +import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.TimeoutException; +import com.github.jobservice.scheduledexecutor.batching.PayloadBatchingService; +import com.github.jobservice.scheduledexecutor.batching.SubdocumentBatchSplitter; +import com.github.jobservice.scheduledexecutor.batching.SubtaskIdGenerator; + /** * This class is responsible for sending task data to the target queue. */ @@ -76,6 +81,21 @@ public QueueServices( public void sendMessage( final String partitionId, final String jobId, final WorkerAction workerAction ) throws IOException, URISyntaxException, InterruptedException, TimeoutException + { + // Check if payload batching is required + if (PayloadBatchingService.shouldBatchPayload(workerAction)) { + sendBatchedMessages(partitionId, jobId, workerAction); + } else { + sendSingleMessage(partitionId, jobId, workerAction); + } + } + + /** + * Sends task data as a single message to the target queue. + */ + private void sendSingleMessage( + final String partitionId, final String jobId, final WorkerAction workerAction + ) throws IOException, URISyntaxException, InterruptedException, TimeoutException { // Generate a random task id. LOG.debug("Generating task id ..."); @@ -95,10 +115,132 @@ public void sendMessage( getStatusCheckIntervalMillis(ScheduledExecutorConfig.getStatusCheckIntervalSeconds()), statusCheckUrl, ScheduledExecutorConfig.getTrackingPipe(), workerAction.getTargetPipe()); - final Object taskMessage = getTaskMessage(trackingInfo, workerAction, taskId); + // Strip the batcher prefix from task pipe if present + final WorkerAction NonBatchingWorkerAction; + if (PayloadBatchingService.hasSubdocumentBatcherPrefix(workerAction)) { + final String strippedTaskPipe = PayloadBatchingService.stripBatcherPrefix(workerAction); + NonBatchingWorkerAction = new WorkerAction(); + NonBatchingWorkerAction.setTaskClassifier(workerAction.getTaskClassifier()); + NonBatchingWorkerAction.setTaskApiVersion(workerAction.getTaskApiVersion()); + NonBatchingWorkerAction.setTaskData(workerAction.getTaskData()); + NonBatchingWorkerAction.setTaskDataEncoding(workerAction.getTaskDataEncoding()); + NonBatchingWorkerAction.setTaskPipe(strippedTaskPipe); + NonBatchingWorkerAction.setTargetPipe(workerAction.getTargetPipe()); + NonBatchingWorkerAction.setCorrelationId(workerAction.getCorrelationId()); + } else { + NonBatchingWorkerAction = workerAction; + } + + final Object taskMessage = getTaskMessage(trackingInfo, NonBatchingWorkerAction, taskId); + + // Serialise and publish + publishTaskMessage(taskMessage); + } + + /** + * Sends task data as multiple batched messages for large DocumentWorkerTask payloads. + */ + private void sendBatchedMessages( + final String partitionId, + final String jobId, + final WorkerAction workerAction + ) throws IOException, URISyntaxException, InterruptedException, TimeoutException + { + // Deserialize taskData JSON to Map + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(workerAction); + if (taskDataMap == null) { + throw new RuntimeException("Failed to deserialize taskData for batching"); + } + + // Extract subdocuments reference + final List subdocuments = SubdocumentBatchSplitter.extractSubdocuments(taskDataMap); + if (subdocuments == null) { + throw new RuntimeException("No subdocuments found for batching"); + } + + final int batchSize = PayloadBatchingService.getBatchSize(); + final int totalBatches = SubdocumentBatchSplitter.calculateBatchCount(subdocuments.size(), batchSize); + + // Generate ONE base task UUID for all batches of this job + final String baseTaskId = UUID.randomUUID().toString(); + final String baseJobTaskId = new JobTaskId(partitionId, jobId).getMessageId(); + + // Status check URL is same for all batches (points to parent job) + final String statusCheckUrl = UriBuilder.fromUri(ScheduledExecutorConfig.getWebserviceUrl()) + .path("partitions").path(partitionId) + .path("jobs").path(jobId) + .path("status").build().toString(); + + // Strip the DocumentWorkerSubdocumentBatcher() prefix from task pipe + final String strippedTaskPipe = PayloadBatchingService.stripBatcherPrefix(workerAction); + + LOG.info("Sending {} batched messages for job {} ({} subdocuments, batch size {}, task pipe: {})", + totalBatches, jobId, subdocuments.size(), batchSize, strippedTaskPipe); + + // Process ONE batch at a time + for (int batchIndex = 1; batchIndex <= totalBatches; batchIndex++) { + + // 1. Get subdocuments subList VIEW for this batch + final List subdocBatch = SubdocumentBatchSplitter.getSubdocumentsBatchView( + subdocuments, batchIndex, batchSize); + + // 2. Create batched taskData Map with this batch's subdocuments + final Map batchedTaskDataMap = SubdocumentBatchSplitter.createBatchedTaskData( + taskDataMap, subdocBatch); + + // 3. Serialize batched taskData to JSON String for WorkerAction + final byte[] batchedTaskDataBytes; + try { + batchedTaskDataBytes = codec.serialise(batchedTaskDataMap); + } catch (final CodecException e) { + throw new RuntimeException("Failed to serialize batched taskData", e); + } + final String batchedTaskDataString = new String(batchedTaskDataBytes, StandardCharsets.UTF_8); + + // 4. Generate subtask IDs for this batch + final String taskSubtaskId = SubtaskIdGenerator.generateTaskSubtaskId( + baseTaskId, batchIndex, totalBatches); + final String jobTaskSubtaskId = SubtaskIdGenerator.generateJobTaskSubtaskId( + baseJobTaskId, batchIndex, totalBatches); - // Serialise the task message. - // Wrap any CodecException as a RuntimeException as it shouldn't happen + // 5. Create TrackingInfo with subtask jobTaskId + final TrackingInfo trackingInfo = new TrackingInfo( + jobTaskSubtaskId, + new Date(), + getStatusCheckIntervalMillis(ScheduledExecutorConfig.getStatusCheckIntervalSeconds()), + statusCheckUrl, + ScheduledExecutorConfig.getTrackingPipe(), + workerAction.getTargetPipe()); + + // 6. Create a modified WorkerAction with batched taskData and stripped task pipe + final WorkerAction batchWorkerAction = new WorkerAction(); + batchWorkerAction.setTaskClassifier(workerAction.getTaskClassifier()); + batchWorkerAction.setTaskApiVersion(workerAction.getTaskApiVersion()); + batchWorkerAction.setTaskData(batchedTaskDataString); + batchWorkerAction.setTaskDataEncoding(WorkerAction.TaskDataEncodingEnum.UTF8); + batchWorkerAction.setTaskPipe(strippedTaskPipe); + batchWorkerAction.setTargetPipe(workerAction.getTargetPipe()); + batchWorkerAction.setCorrelationId(workerAction.getCorrelationId()); + + // 7. Reuse existing getTaskMessage() method + final TaskMessage taskMessage = getTaskMessage(trackingInfo, batchWorkerAction, taskSubtaskId); + + // 8. Publish this batch + LOG.debug("Sending batch {}/{} with {} subdocuments (taskId={}, jobTaskId={})", + batchIndex, totalBatches, subdocBatch.size(), taskSubtaskId, jobTaskSubtaskId); + + publishTaskMessage(taskMessage); + } + + LOG.info("Successfully sent all {} batches for job {}", totalBatches, jobId); + } + + /** + * Publishes a task message to RabbitMQ. + */ + private void publishTaskMessage(final Object taskMessage) + throws IOException, InterruptedException, TimeoutException + { final byte[] taskMessageBytes; try { LOG.debug("Serialise the task message ..."); diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java new file mode 100644 index 000000000..1672386ed --- /dev/null +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java @@ -0,0 +1,194 @@ +/* + * Copyright 2016-2026 Open Text. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.jobservice.scheduledexecutor.batching; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.jobservice.scheduledexecutor.WorkerAction; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Map; + +/** + * Service responsible for detecting when payload batching is needed. + *

+ * This service implements the payload batching strategy for DocumentWorkerTask jobs + * with large subdocuments arrays, splitting them into smaller batches. + *

+ * Batching is opt-in: only jobs whose task pipe starts with the + * {@code DocumentWorkerSubdocumentBatcher()} prefix will be batched. + */ +public final class PayloadBatchingService +{ + private static final Logger LOG = LoggerFactory.getLogger(PayloadBatchingService.class); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final TypeReference> MAP_TYPE_REF = new TypeReference<>() {}; + + /** Task classifier that triggers batching */ + public static final String DOCUMENT_WORKER_TASK_CLASSIFIER = "DocumentWorkerTask"; + + /** Task pipe prefix that opts-in to subdocument batching */ + public static final String SUBDOCUMENT_BATCHER_PREFIX = "DocumentWorkerSubdocumentBatcher() "; + + /** Maximum number of subdocuments per batch */ + public static final int BATCH_SIZE = 200; + + private PayloadBatchingService() + { + } + + /** + * Determines whether a worker action's payload should be batched. + *

+ * Batching is triggered only when ALL of these conditions are met: + *

    + *
  • taskClassifier == "DocumentWorkerTask"
    • + *
  • task pipe starts with "DocumentWorkerSubdocumentBatcher() "
    • + *
  • payload has document.subdocuments array
    • + *
  • subdocuments array size > BATCH_SIZE
    • + *
+ * + * @param workerAction The worker action to evaluate + * @return true if the payload should be batched, false otherwise + */ + public static boolean shouldBatchPayload(final WorkerAction workerAction) + { + if (!DOCUMENT_WORKER_TASK_CLASSIFIER.equals(workerAction.getTaskClassifier())) { + LOG.debug("Not a DocumentWorkerTask, skipping batching"); + return false; + } + + if (!hasSubdocumentBatcherPrefix(workerAction)) { + LOG.debug("Task pipe does not have DocumentWorkerSubdocumentBatcher() prefix, skipping batching"); + return false; + } + + final int subdocCount = getSubdocumentsCount(workerAction); + if (subdocCount < 0) { + LOG.debug("Could not determine subdocuments count, skipping batching"); + return false; + } + + if (subdocCount <= BATCH_SIZE) { + LOG.debug("Subdocuments count ({}) <= batch size ({}), skipping batching", + subdocCount, BATCH_SIZE); + return false; + } + + LOG.info("Payload batching required: {} subdocuments will be split into batches of {}", + subdocCount, BATCH_SIZE); + return true; + } + + /** + * Checks if the task pipe starts with the DocumentWorkerSubdocumentBatcher() prefix. + * @param workerAction The worker action + * @return true if task pipe has the batching prefix, false otherwise + */ + public static boolean hasSubdocumentBatcherPrefix(final WorkerAction workerAction) + { + final String taskPipe = workerAction.getTaskPipe(); + return taskPipe != null && taskPipe.startsWith(SUBDOCUMENT_BATCHER_PREFIX); + } + + /** + * Strips the DocumentWorkerSubdocumentBatcher() prefix from the task pipe. + * @param workerAction The worker action + * @return The task pipe without the prefix, or the original task pipe if prefix not present + */ + public static String stripBatcherPrefix(final WorkerAction workerAction) + { + final String taskPipe = workerAction.getTaskPipe(); + if (taskPipe != null && taskPipe.startsWith(SUBDOCUMENT_BATCHER_PREFIX)) { + return taskPipe.substring(SUBDOCUMENT_BATCHER_PREFIX.length()); + } + return taskPipe; + } + + /** + * Gets the subdocuments count from a worker action. + * @param workerAction The worker action + * @return Subdocuments count, or -1 if cannot be determined + */ + public static int getSubdocumentsCount(final WorkerAction workerAction) + { + final Map taskDataMap = deserializeTaskData(workerAction); + if (taskDataMap == null) { + return -1; + } + + final List subdocuments = SubdocumentBatchSplitter.extractSubdocuments(taskDataMap); + if (subdocuments == null) { + return -1; + } + + return subdocuments.size(); + } + + /** + * Deserializes the taskData JSON String to a Map. + * @param workerAction The worker action containing taskData + * @return Deserialized Map, or null if deserialization fails + */ + public static Map deserializeTaskData(final WorkerAction workerAction) + { + final Object taskDataObj = workerAction.getTaskData(); + + if (taskDataObj == null) { + LOG.debug("TaskData is null"); + return null; + } + + if (!(taskDataObj instanceof String)) { + LOG.debug("TaskData is not a String, got: {}", taskDataObj.getClass().getName()); + return null; + } + + try { + return OBJECT_MAPPER.readValue((String) taskDataObj, MAP_TYPE_REF); + } catch (final JsonProcessingException e) { + LOG.debug("Failed to deserialize taskData as JSON: {}", e.getMessage()); + return null; + } + } + + /** + * Calculates the total number of batches needed. + * @param workerAction The worker action + * @return Number of batches, or 0 if batching not applicable + */ + public static int calculateTotalBatches(final WorkerAction workerAction) + { + final int subdocCount = getSubdocumentsCount(workerAction); + if (subdocCount <= 0) { + return 0; + } + return SubdocumentBatchSplitter.calculateBatchCount(subdocCount, BATCH_SIZE); + } + + /** + * Gets the batch size constant. + * @return The batch size (200) + */ + public static int getBatchSize() + { + return BATCH_SIZE; + } +} + diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java new file mode 100644 index 000000000..d72e0cf9e --- /dev/null +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java @@ -0,0 +1,152 @@ +/* + * Copyright 2016-2026 Open Text. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.jobservice.scheduledexecutor.batching; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Utility class for subdocuments extraction and batching. + *

+ * This class uses index-based processing to avoid creating copies of large + * subdocument arrays. Instead of splitting the entire list upfront, it provides + * methods to get batch ranges and create batched payloads one at a time. + */ +public final class SubdocumentBatchSplitter +{ + /** JSON key for the document object in taskData */ + public static final String DOCUMENT_KEY = "document"; + + /** JSON key for the subdocuments array within document */ + public static final String SUBDOCUMENTS_KEY = "subdocuments"; + + private SubdocumentBatchSplitter() + { + } + + /** + * Extracts the subdocuments list from a taskData map. + * @param taskData The taskData map (deserialized from JSON) + * @return The subdocuments list, or null if not found or not a list + */ + @SuppressWarnings("unchecked") + public static List extractSubdocuments(final Map taskData) + { + if (taskData == null) { + return null; + } + + final Object documentObj = taskData.get(DOCUMENT_KEY); + if (!(documentObj instanceof Map)) { + return null; + } + + final Map document = (Map) documentObj; + final Object subdocsObj = document.get(SUBDOCUMENTS_KEY); + + if (!(subdocsObj instanceof List)) { + return null; + } + + return (List) subdocsObj; + } + + /** + * Calculates the number of batches needed for a given item count and batch size. + * @param totalItems Total number of items + * @param batchSize Maximum items per batch + * @return Number of batches required + */ + public static int calculateBatchCount(final int totalItems, final int batchSize) + { + if (totalItems <= 0) { + return 0; + } + if (batchSize < 1) { + throw new IllegalArgumentException("Batch size must be >= 1"); + } + return (int) Math.ceil((double) totalItems / batchSize); + } + + /** + * Gets the start index (inclusive) for a specific batch. + * @param batchIndex 1-based batch index + * @param batchSize Items per batch + * @return Start index (0-based, inclusive) + */ + public static int getBatchStartIndex(final int batchIndex, final int batchSize) + { + if (batchIndex < 1) { + throw new IllegalArgumentException("Batch index must be >= 1"); + } + return (batchIndex - 1) * batchSize; + } + + /** + * Gets the end index (exclusive) for a specific batch. + * @param batchIndex 1-based batch index + * @param batchSize Items per batch + * @param totalItems Total number of items + * @return End index (0-based, exclusive) + */ + public static int getBatchEndIndex(final int batchIndex, final int batchSize, final int totalItems) + { + final int start = getBatchStartIndex(batchIndex, batchSize); + return Math.min(start + batchSize, totalItems); + } + + /** + * Gets the subdocuments for a specific batch as a subList VIEW. + * @param subdocuments The full subdocuments list + * @param batchIndex 1-based batch index + * @param batchSize Items per batch + * @return SubList view of subdocuments for this batch + */ + public static List getSubdocumentsBatchView( + final List subdocuments, + final int batchIndex, + final int batchSize) + { + final int start = getBatchStartIndex(batchIndex, batchSize); + final int end = getBatchEndIndex(batchIndex, batchSize, subdocuments.size()); + return subdocuments.subList(start, end); + } + + /** + * Creates a new taskData map with a batch of subdocuments replacing the original. + * @param originalTaskData The original taskData map + * @param subdocumentsBatch The batch of subdocuments to use (can be a subList view) + * @return New taskData map with batched subdocuments + */ + @SuppressWarnings("unchecked") + public static Map createBatchedTaskData( + final Map originalTaskData, + final List subdocumentsBatch) + { + final Map batchedTaskData = new HashMap<>(originalTaskData); + final Map originalDocument = (Map) originalTaskData.get(DOCUMENT_KEY); + + if (originalDocument != null) { + final Map batchedDocument = new HashMap<>(originalDocument); + batchedDocument.put(SUBDOCUMENTS_KEY, subdocumentsBatch); + batchedTaskData.put(DOCUMENT_KEY, batchedDocument); + } + + return batchedTaskData; + } +} + diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubtaskIdGenerator.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubtaskIdGenerator.java new file mode 100644 index 000000000..871cc1e80 --- /dev/null +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubtaskIdGenerator.java @@ -0,0 +1,96 @@ +/* + * Copyright 2016-2026 Open Text. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.jobservice.scheduledexecutor.batching; + +/** + * Generates subtask IDs for batched messages + *

+ * Subtask IDs follow the format: {baseId}.{subtaskNumber} or {baseId}.{subtaskNumber}* for the final subtask. + * The asterisk (*) marker on the final subtask is critical for progress calculation. + */ +public final class SubtaskIdGenerator +{ + private SubtaskIdGenerator() + { + } + + /** + * Generates a subtask ID for a TaskMessage. + * @param baseTaskId The base task UUID + * @param subtaskIndex The 1-based index of this subtask + * @param totalSubtasks The total number of subtasks + * @return The subtask ID + */ + public static String generateTaskSubtaskId( + final String baseTaskId, + final int subtaskIndex, + final int totalSubtasks) + { + validateInputs(subtaskIndex, totalSubtasks); + return formatSubtaskId(baseTaskId, subtaskIndex, subtaskIndex == totalSubtasks); + } + + /** + * Generates a subtask ID for TrackingInfo.jobTaskId. + * @param baseJobTaskId The base job task ID + * @param subtaskIndex The 1-based index of this subtask + * @param totalSubtasks The total number of subtasks + * @return The subtask job task ID + */ + public static String generateJobTaskSubtaskId( + final String baseJobTaskId, + final int subtaskIndex, + final int totalSubtasks) + { + validateInputs(subtaskIndex, totalSubtasks); + return formatSubtaskId(baseJobTaskId, subtaskIndex, subtaskIndex == totalSubtasks); + } + + /** + * Formats the subtask ID with optional final marker. + */ + private static String formatSubtaskId( + final String baseId, + final int subtaskIndex, + final boolean isFinal) + { + final StringBuilder builder = new StringBuilder(baseId); + builder.append('.'); + builder.append(subtaskIndex); + if (isFinal) { + builder.append('*'); + } + return builder.toString(); + } + + /** + * Validates subtask index inputs. + */ + private static void validateInputs(final int subtaskIndex, final int totalSubtasks) + { + if (subtaskIndex < 1) { + throw new IllegalArgumentException("Subtask index must be >= 1, got: " + subtaskIndex); + } + if (totalSubtasks < 1) { + throw new IllegalArgumentException("Total subtasks must be >= 1, got: " + totalSubtasks); + } + if (subtaskIndex > totalSubtasks) { + throw new IllegalArgumentException( + "Subtask index (" + subtaskIndex + ") cannot exceed total (" + totalSubtasks + ")"); + } + } +} + diff --git a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java new file mode 100644 index 000000000..ee50112d7 --- /dev/null +++ b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java @@ -0,0 +1,523 @@ +/* + * Copyright 2016-2026 Open Text. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.jobservice.scheduledexecutor.batching; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.jobservice.scheduledexecutor.WorkerAction; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Unit tests for payload batching classes: PayloadBatchingService, SubdocumentBatchSplitter, SubtaskIdGenerator. + */ +public class PayloadBatchingTest +{ + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final int BATCH_SIZE = PayloadBatchingService.BATCH_SIZE; + private static final String SUBDOCUMENT_BATCHER_PREFIX = PayloadBatchingService.SUBDOCUMENT_BATCHER_PREFIX; + private static final String PARTITION_ID = "tenant-user1"; + private static final String JOB_ID = "job123"; + private static final String TARGET_PIPE = "target-pipe"; + private static final String BASE_TASK_PIPE = "worker-queue"; + private static final String TASK_PIPE_WITH_PREFIX = SUBDOCUMENT_BATCHER_PREFIX + BASE_TASK_PIPE; + + // ===================================================== + // DETECTION TESTS (PayloadBatchingService) + // Opt-in via task pipe prefix: DocumentWorkerSubdocumentBatcher() + // ===================================================== + + @Test + public void testDocumentWorkerTask_WithPrefix_NoSubdocuments_NoSplitting() throws JsonProcessingException + { + // DocumentWorkerTask with prefix but without subdocuments field + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataWithoutSubdocs()); + assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + } + + @Test + public void testDocumentWorkerTask_WithPrefix_EmptySubdocuments_NoSplitting() throws JsonProcessingException + { + // DocumentWorkerTask with prefix and empty subdocuments array + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(0)); + assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + } + + @Test + public void testDocumentWorkerTask_WithPrefix_SmallSubdocuments_NoSplitting() throws JsonProcessingException + { + // DocumentWorkerTask with prefix and subdocuments below threshold + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(BATCH_SIZE / 2)); + assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + } + + @Test + public void testDocumentWorkerTask_WithPrefix_LargeSubdocuments_SplittingEnabled() throws JsonProcessingException + { + // DocumentWorkerTask with prefix and subdocuments above threshold + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(BATCH_SIZE + 1)); + assertTrue(PayloadBatchingService.shouldBatchPayload(action)); + } + + @Test + public void testDocumentWorkerTask_NoPrefix_LargeSubdocuments_NoSplitting() throws JsonProcessingException + { + // DocumentWorkerTask WITHOUT prefix but with large subdocuments (not opted in) + final WorkerAction action = createWorkerActionWithoutPrefix(createTaskDataMap(BATCH_SIZE * 5)); + assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + } + + @Test + public void testNonDocumentWorkerTask_WithPrefix_NoSplitting() throws JsonProcessingException + { + // Non-DocumentWorkerTask with prefix in task pipe (wrong classifier) + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(BATCH_SIZE * 5)); + action.setTaskClassifier("NonDocumentWorkerTask"); + assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + } + + @Test + public void testShouldBatch_NullTaskData_ReturnsFalse() + { + final WorkerAction action = new WorkerAction(); + action.setTaskClassifier(PayloadBatchingService.DOCUMENT_WORKER_TASK_CLASSIFIER); + action.setTaskPipe(TASK_PIPE_WITH_PREFIX); + action.setTaskData(null); + assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + } + + // ===================================================== + // TASK PIPE PREFIX TESTS + // ===================================================== + + @Test + public void testHasSubdocumentBatcherPrefix_WithPrefix() + { + final WorkerAction action = new WorkerAction(); + action.setTaskPipe(TASK_PIPE_WITH_PREFIX); + assertTrue(PayloadBatchingService.hasSubdocumentBatcherPrefix(action)); + } + + @Test + public void testHasSubdocumentBatcherPrefix_WithoutPrefix() + { + final WorkerAction action = new WorkerAction(); + action.setTaskPipe(BASE_TASK_PIPE); + assertFalse(PayloadBatchingService.hasSubdocumentBatcherPrefix(action)); + } + + @Test + public void testHasSubdocumentBatcherPrefix_NullTaskPipe() + { + final WorkerAction action = new WorkerAction(); + action.setTaskPipe(null); + assertFalse(PayloadBatchingService.hasSubdocumentBatcherPrefix(action)); + } + + @Test + public void testStripBatcherPrefix_WithPrefix() + { + final WorkerAction action = new WorkerAction(); + action.setTaskPipe(TASK_PIPE_WITH_PREFIX); + assertEquals(BASE_TASK_PIPE, PayloadBatchingService.stripBatcherPrefix(action)); + } + + @Test + public void testStripBatcherPrefix_WithoutPrefix() + { + final WorkerAction action = new WorkerAction(); + action.setTaskPipe(BASE_TASK_PIPE); + assertEquals(BASE_TASK_PIPE, PayloadBatchingService.stripBatcherPrefix(action)); + } + + @Test + public void testStripBatcherPrefix_NullTaskPipe() + { + final WorkerAction action = new WorkerAction(); + action.setTaskPipe(null); + assertNull(PayloadBatchingService.stripBatcherPrefix(action)); + } + + @Test + public void testGetSubdocumentsCount_Valid() throws JsonProcessingException + { + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(500)); + assertEquals(500, PayloadBatchingService.getSubdocumentsCount(action)); + } + + @Test + public void testCalculateTotalBatches() throws JsonProcessingException + { + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(500)); + assertEquals(3, PayloadBatchingService.calculateTotalBatches(action)); // 500/200 = 2.5 = 3 + } + + @Test + public void testDeserializeTaskData_Valid() throws JsonProcessingException + { + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(10)); + final Map result = PayloadBatchingService.deserializeTaskData(action); + assertNotNull(result); + assertTrue(result.containsKey("document")); + } + + @Test + public void testDeserializeTaskData_InvalidJson() + { + final WorkerAction action = new WorkerAction(); + action.setTaskClassifier(PayloadBatchingService.DOCUMENT_WORKER_TASK_CLASSIFIER); + action.setTaskPipe(TASK_PIPE_WITH_PREFIX); + action.setTaskData("invalid json {{{"); + assertNull(PayloadBatchingService.deserializeTaskData(action)); + } + + // ===================================================== + // SPLITTING LOGIC TESTS (SubdocumentBatchSplitter) + // ===================================================== + + @Test + public void testCalculateBatchCount_ExactDivision() + { + assertEquals(10, SubdocumentBatchSplitter.calculateBatchCount(1000, 100)); + } + + @Test + public void testCalculateBatchCount_UnevenDivision() + { + assertEquals(11, SubdocumentBatchSplitter.calculateBatchCount(1050, 100)); + } + + @Test + public void testCalculateBatchCount_JustOverThreshold() + { + assertEquals(2, SubdocumentBatchSplitter.calculateBatchCount(BATCH_SIZE + 1, BATCH_SIZE)); + } + + @Test + public void testCalculateBatchCount_ZeroItems() + { + assertEquals(0, SubdocumentBatchSplitter.calculateBatchCount(0, 100)); + } + + @Test + public void testCalculateBatchCount_InvalidBatchSize() + { + assertThrows(IllegalArgumentException.class, + () -> SubdocumentBatchSplitter.calculateBatchCount(100, 0)); + } + + @Test + public void testGetBatchStartIndex() + { + assertEquals(0, SubdocumentBatchSplitter.getBatchStartIndex(1, 100)); + assertEquals(100, SubdocumentBatchSplitter.getBatchStartIndex(2, 100)); + assertEquals(200, SubdocumentBatchSplitter.getBatchStartIndex(3, 100)); + } + + @Test + public void testGetBatchStartIndex_InvalidIndex() + { + assertThrows(IllegalArgumentException.class, + () -> SubdocumentBatchSplitter.getBatchStartIndex(0, 100)); + } + + @Test + public void testGetSubdocumentsBatchView_PreservesOrder() + { + final List subdocs = createSubdocList(500); + final int batchSize = 100; + + final List batch1 = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, 1, batchSize); + final List batch3 = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, 3, batchSize); + + assertEquals("1", getSubdocId(batch1.get(0))); + assertEquals("100", getSubdocId(batch1.get(99))); + assertEquals("201", getSubdocId(batch3.get(0))); + assertEquals("300", getSubdocId(batch3.get(99))); + } + + @Test + public void testGetSubdocumentsBatchView_LastBatch() + { + final List subdocs = createSubdocList(250); + final int batchSize = 100; + + final List lastBatch = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, 3, batchSize); + assertEquals(50, lastBatch.size()); + assertEquals("201", getSubdocId(lastBatch.get(0))); + assertEquals("250", getSubdocId(lastBatch.get(49))); + } + + @Test + public void testExtractSubdocuments_MissingDocument() + { + final Map taskData = new HashMap<>(); + taskData.put("someField", "value"); + assertNull(SubdocumentBatchSplitter.extractSubdocuments(taskData)); + } + + @Test + public void testExtractSubdocuments_MissingSubdocuments() + { + final Map taskData = new HashMap<>(); + taskData.put("document", new HashMap<>()); + assertNull(SubdocumentBatchSplitter.extractSubdocuments(taskData)); + } + + // ===================================================== + // PAYLOAD INTEGRITY TESTS + // ===================================================== + + @Test + @SuppressWarnings("unchecked") + public void testCreateBatchedTaskData_PreservesOtherFields() + { + final Map original = createTaskDataMap(500); + final List subdocs = SubdocumentBatchSplitter.extractSubdocuments(original); + + for (int i = 1; i <= 3; i++) { + final List batch = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, i, BATCH_SIZE); + final Map batched = SubdocumentBatchSplitter.createBatchedTaskData(original, batch); + + // Verify customData and scripts preserved + assertEquals(original.get("customData"), batched.get("customData")); + assertEquals(original.get("scripts"), batched.get("scripts")); + + // Verify document fields are preserved in each batch + final Map originalDoc = (Map) original.get("document"); + final Map batchedDoc = (Map) batched.get("document"); + assertEquals(originalDoc.get("fields"), batchedDoc.get("fields"), + "Document fields should be preserved in batch " + i); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testCreateBatchedTaskData_CorrectSubdocumentsInEachBatch() + { + final Map original = createTaskDataMap(500); + final List subdocs = SubdocumentBatchSplitter.extractSubdocuments(original); + + final List batch1 = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, 1, BATCH_SIZE); + final List batch2 = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, 2, BATCH_SIZE); + final List batch3 = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, 3, BATCH_SIZE); + + assertEquals(BATCH_SIZE, batch1.size()); + assertEquals(BATCH_SIZE, batch2.size()); + assertEquals(100, batch3.size()); + + assertEquals("1", getSubdocId(batch1.get(0))); + assertEquals("200", getSubdocId(batch1.get(BATCH_SIZE - 1))); + assertEquals("201", getSubdocId(batch2.get(0))); + assertEquals("400", getSubdocId(batch2.get(BATCH_SIZE - 1))); + assertEquals("401", getSubdocId(batch3.get(0))); + assertEquals("500", getSubdocId(batch3.get(batch3.size() - 1))); + } + + @Test + public void testBatching_NoSubdocumentDuplication() + { + final List subdocs = createSubdocList(500); + final Set seen = new HashSet<>(); + final int totalBatches = SubdocumentBatchSplitter.calculateBatchCount(500, BATCH_SIZE); + + for (int i = 1; i <= totalBatches; i++) { + for (final Object item : SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, i, BATCH_SIZE)) { + final String id = getSubdocId(item); + assertFalse(seen.contains(id), "Duplicate: " + id); + seen.add(id); + } + } + assertEquals(500, seen.size()); + } + + @Test + public void testBatching_NoSubdocumentLoss() + { + final List subdocs = createSubdocList(500); + int total = 0; + final int totalBatches = SubdocumentBatchSplitter.calculateBatchCount(500, BATCH_SIZE); + + for (int i = 1; i <= totalBatches; i++) { + total += SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, i, BATCH_SIZE).size(); + } + assertEquals(500, total); + } + + // ===================================================== + // SUBTASK ID TESTS (SubtaskIdGenerator) + // ===================================================== + + @Test + public void testGenerateJobTaskSubtaskId_Format() + { + final String baseJobTaskId = PARTITION_ID + ":" + JOB_ID; + assertEquals(PARTITION_ID + ":" + JOB_ID + ".1", + SubtaskIdGenerator.generateJobTaskSubtaskId(baseJobTaskId, 1, 3)); + assertEquals(PARTITION_ID + ":" + JOB_ID + ".2", + SubtaskIdGenerator.generateJobTaskSubtaskId(baseJobTaskId, 2, 3)); + assertEquals(PARTITION_ID + ":" + JOB_ID + ".3*", + SubtaskIdGenerator.generateJobTaskSubtaskId(baseJobTaskId, 3, 3)); + } + + @Test + public void testGenerateTaskSubtaskId_Format() + { + final String baseTaskId = "abc-123-def-456"; + assertEquals("abc-123-def-456.1", SubtaskIdGenerator.generateTaskSubtaskId(baseTaskId, 1, 3)); + assertEquals("abc-123-def-456.2", SubtaskIdGenerator.generateTaskSubtaskId(baseTaskId, 2, 3)); + assertEquals("abc-123-def-456.3*", SubtaskIdGenerator.generateTaskSubtaskId(baseTaskId, 3, 3)); + } + + @Test + public void testSubtaskId_OnlyLastHasFinalMarker() + { + final String baseId = "base"; + for (int i = 1; i < 5; i++) { + assertFalse(SubtaskIdGenerator.generateTaskSubtaskId(baseId, i, 5).endsWith("*")); + } + assertTrue(SubtaskIdGenerator.generateTaskSubtaskId(baseId, 5, 5).endsWith("*")); + } + + @Test + public void testSubtaskId_SingleBatch_HasFinalMarker() + { + assertEquals("base.1*", SubtaskIdGenerator.generateTaskSubtaskId("base", 1, 1)); + } + + @Test + public void testSubtaskId_BothIdsHaveMatchingSuffix() + { + final String baseTaskId = "task-uuid"; + final String baseJobTaskId = "partition:job"; + + for (int i = 1; i <= 3; i++) { + final String taskSuffix = SubtaskIdGenerator.generateTaskSubtaskId(baseTaskId, i, 3) + .substring(baseTaskId.length()); + final String jobTaskSuffix = SubtaskIdGenerator.generateJobTaskSubtaskId(baseJobTaskId, i, 3) + .substring(baseJobTaskId.length()); + assertEquals(taskSuffix, jobTaskSuffix); + } + } + + @Test + public void testSubtaskId_InvalidSubtaskIndex() + { + assertThrows(IllegalArgumentException.class, + () -> SubtaskIdGenerator.generateTaskSubtaskId("base", 0, 3)); + assertThrows(IllegalArgumentException.class, + () -> SubtaskIdGenerator.generateTaskSubtaskId("base", 4, 3)); + } + + // ===================================================== + // HELPER METHODS + // ===================================================== + + /** + * Creates a WorkerAction WITH the DocumentWorkerSubdocumentBatcher() prefix (opted-in for batching) + */ + private WorkerAction createWorkerActionWithPrefix(final Map taskData) throws JsonProcessingException + { + final WorkerAction action = new WorkerAction(); + action.setTaskClassifier(PayloadBatchingService.DOCUMENT_WORKER_TASK_CLASSIFIER); + action.setTaskApiVersion(2); + action.setTaskData(OBJECT_MAPPER.writeValueAsString(taskData)); + action.setTaskPipe(TASK_PIPE_WITH_PREFIX); + action.setTargetPipe(TARGET_PIPE); + return action; + } + + /** + * Creates a WorkerAction WITHOUT the DocumentWorkerSubdocumentBatcher() prefix (not opted-in) + */ + private WorkerAction createWorkerActionWithoutPrefix(final Map taskData) throws JsonProcessingException + { + final WorkerAction action = new WorkerAction(); + action.setTaskClassifier(PayloadBatchingService.DOCUMENT_WORKER_TASK_CLASSIFIER); + action.setTaskApiVersion(2); + action.setTaskData(OBJECT_MAPPER.writeValueAsString(taskData)); + action.setTaskPipe(BASE_TASK_PIPE); + action.setTargetPipe(TARGET_PIPE); + return action; + } + + // Creates taskData with actual structure - numbering starts from 1 + private Map createTaskDataMap(final int subdocCount) + { + final Map taskData = new HashMap<>(); + + final Map document = new HashMap<>(); + // Add fields inside document (like targetId, destinationId in real payloads) + final Map documentFields = new HashMap<>(); + documentFields.put("targetId", List.of(Map.of("data", "2"))); + documentFields.put("destinationId", List.of(Map.of("data", "4"))); + document.put("fields", documentFields); + document.put("subdocuments", createSubdocList(subdocCount)); + taskData.put("document", document); + + final Map customData = new HashMap<>(); + customData.put("tenantId", "testTenant"); + customData.put("workflowName", "test-workflow"); + customData.put("timestamp", "1774593207"); + taskData.put("customData", customData); + + taskData.put("scripts", new ArrayList<>()); + + return taskData; + } + + // Creates subdoc list with actual structure - numbering starts from 1 + private List createSubdocList(final int count) + { + final List subdocs = new ArrayList<>(); + for (int i = 1; i <= count; i++) { + final Map subdoc = new HashMap<>(); + final Map fields = new HashMap<>(); + fields.put("_id", List.of(Map.of("data", String.valueOf(i)))); + fields.put("_index", List.of(Map.of("data", "test-index"))); + fields.put("_routing", List.of(Map.of("data", String.valueOf(i)))); + subdoc.put("fields", fields); + subdocs.add(subdoc); + } + return subdocs; + } + + private Map createTaskDataWithoutSubdocs() + { + final Map taskData = new HashMap<>(); + taskData.put("document", new HashMap<>()); + taskData.put("customData", Map.of("key", "value")); + taskData.put("scripts", new ArrayList<>()); + return taskData; + } + + @SuppressWarnings("unchecked") + private String getSubdocId(final Object subdoc) + { + final Map subdocMap = (Map) subdoc; + final Map fields = (Map) subdocMap.get("fields"); + final List> idList = (List>) fields.get("_id"); + return (String) idList.get(0).get("data"); + } +} From 53db620074c81f853ea515a3d7450266112b2a33 Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Thu, 14 May 2026 11:36:59 +0530 Subject: [PATCH 2/9] US1124088: Payload Batching job-service enhancement --- .../scheduledexecutor/DatabasePoller.java | 6 +- .../scheduledexecutor/QueueServices.java | 7 +- .../batching/PayloadBatchingService.java | 29 +--- .../batching/PayloadBatchingTest.java | 56 +------- .../github/jobservice/util/TaskPipeUtil.java | 64 +++++++++ .../jobservice/util/TaskPipeUtilTest.java | 125 ++++++++++++++++++ .../jobtracking/JobTrackingWorkerUtil.java | 6 +- 7 files changed, 207 insertions(+), 86 deletions(-) create mode 100644 job-service-util/src/main/java/com/github/jobservice/util/TaskPipeUtil.java create mode 100644 job-service-util/src/test/java/com/github/jobservice/util/TaskPipeUtilTest.java diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java index 680c2b8ee..a9b808f0a 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java @@ -20,6 +20,7 @@ import com.github.cafapi.common.api.Codec; import com.github.cafapi.common.util.moduleloader.ModuleLoader; import com.github.cafapi.common.util.moduleloader.ModuleLoaderException; +import com.github.jobservice.util.TaskPipeUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -88,7 +89,10 @@ private static void sendMessageToQueueMessaging(final Codec codec, final JobTask final String context = MessageFormat.format("[partitionId={0}, jobId={1}, taskPipe={2}]", partitionId, jobId, taskPipe); - try (final QueueServices queueServices = QueueServicesFactory.create(taskPipe, partitionId, codec)) { + // Strip the DocumentWorkerSubdocumentBatcher() prefix from taskPipe to get the actual target queue. + final String actualTargetQueue = TaskPipeUtil.stripBatcherPrefix(workerAction.getTaskPipe()); + + try (final QueueServices queueServices = QueueServicesFactory.create(actualTargetQueue, partitionId, codec)) { queueServices.sendMessage(partitionId, jobId, workerAction); diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java index 5eea0484a..175cd708f 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java @@ -42,6 +42,7 @@ import com.github.jobservice.scheduledexecutor.batching.PayloadBatchingService; import com.github.jobservice.scheduledexecutor.batching.SubdocumentBatchSplitter; import com.github.jobservice.scheduledexecutor.batching.SubtaskIdGenerator; +import com.github.jobservice.util.TaskPipeUtil; /** * This class is responsible for sending task data to the target queue. @@ -117,8 +118,8 @@ private void sendSingleMessage( // Strip the batcher prefix from task pipe if present final WorkerAction NonBatchingWorkerAction; - if (PayloadBatchingService.hasSubdocumentBatcherPrefix(workerAction)) { - final String strippedTaskPipe = PayloadBatchingService.stripBatcherPrefix(workerAction); + if (TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { + final String strippedTaskPipe = TaskPipeUtil.stripBatcherPrefix(workerAction.getTaskPipe()); NonBatchingWorkerAction = new WorkerAction(); NonBatchingWorkerAction.setTaskClassifier(workerAction.getTaskClassifier()); NonBatchingWorkerAction.setTaskApiVersion(workerAction.getTaskApiVersion()); @@ -172,7 +173,7 @@ private void sendBatchedMessages( .path("status").build().toString(); // Strip the DocumentWorkerSubdocumentBatcher() prefix from task pipe - final String strippedTaskPipe = PayloadBatchingService.stripBatcherPrefix(workerAction); + final String strippedTaskPipe = TaskPipeUtil.stripBatcherPrefix(workerAction.getTaskPipe()); LOG.info("Sending {} batched messages for job {} ({} subdocuments, batch size {}, task pipe: {})", totalBatches, jobId, subdocuments.size(), batchSize, strippedTaskPipe); diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java index 1672386ed..a8d8898c1 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java @@ -19,6 +19,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.github.jobservice.scheduledexecutor.WorkerAction; +import com.github.jobservice.util.TaskPipeUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -43,8 +44,6 @@ public final class PayloadBatchingService /** Task classifier that triggers batching */ public static final String DOCUMENT_WORKER_TASK_CLASSIFIER = "DocumentWorkerTask"; - /** Task pipe prefix that opts-in to subdocument batching */ - public static final String SUBDOCUMENT_BATCHER_PREFIX = "DocumentWorkerSubdocumentBatcher() "; /** Maximum number of subdocuments per batch */ public static final int BATCH_SIZE = 200; @@ -74,7 +73,7 @@ public static boolean shouldBatchPayload(final WorkerAction workerAction) return false; } - if (!hasSubdocumentBatcherPrefix(workerAction)) { + if (!TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { LOG.debug("Task pipe does not have DocumentWorkerSubdocumentBatcher() prefix, skipping batching"); return false; } @@ -96,30 +95,6 @@ public static boolean shouldBatchPayload(final WorkerAction workerAction) return true; } - /** - * Checks if the task pipe starts with the DocumentWorkerSubdocumentBatcher() prefix. - * @param workerAction The worker action - * @return true if task pipe has the batching prefix, false otherwise - */ - public static boolean hasSubdocumentBatcherPrefix(final WorkerAction workerAction) - { - final String taskPipe = workerAction.getTaskPipe(); - return taskPipe != null && taskPipe.startsWith(SUBDOCUMENT_BATCHER_PREFIX); - } - - /** - * Strips the DocumentWorkerSubdocumentBatcher() prefix from the task pipe. - * @param workerAction The worker action - * @return The task pipe without the prefix, or the original task pipe if prefix not present - */ - public static String stripBatcherPrefix(final WorkerAction workerAction) - { - final String taskPipe = workerAction.getTaskPipe(); - if (taskPipe != null && taskPipe.startsWith(SUBDOCUMENT_BATCHER_PREFIX)) { - return taskPipe.substring(SUBDOCUMENT_BATCHER_PREFIX.length()); - } - return taskPipe; - } /** * Gets the subdocuments count from a worker action. diff --git a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java index ee50112d7..d22761a1b 100644 --- a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java +++ b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java @@ -18,6 +18,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.github.jobservice.scheduledexecutor.WorkerAction; +import com.github.jobservice.util.TaskPipeUtil; import org.junit.jupiter.api.Test; import java.util.ArrayList; @@ -36,12 +37,11 @@ public class PayloadBatchingTest { private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); private static final int BATCH_SIZE = PayloadBatchingService.BATCH_SIZE; - private static final String SUBDOCUMENT_BATCHER_PREFIX = PayloadBatchingService.SUBDOCUMENT_BATCHER_PREFIX; private static final String PARTITION_ID = "tenant-user1"; private static final String JOB_ID = "job123"; private static final String TARGET_PIPE = "target-pipe"; private static final String BASE_TASK_PIPE = "worker-queue"; - private static final String TASK_PIPE_WITH_PREFIX = SUBDOCUMENT_BATCHER_PREFIX + BASE_TASK_PIPE; + private static final String TASK_PIPE_WITH_PREFIX = TaskPipeUtil.SUBDOCUMENT_BATCHER_PREFIX + BASE_TASK_PIPE; // ===================================================== // DETECTION TESTS (PayloadBatchingService) @@ -107,58 +107,6 @@ public void testShouldBatch_NullTaskData_ReturnsFalse() assertFalse(PayloadBatchingService.shouldBatchPayload(action)); } - // ===================================================== - // TASK PIPE PREFIX TESTS - // ===================================================== - - @Test - public void testHasSubdocumentBatcherPrefix_WithPrefix() - { - final WorkerAction action = new WorkerAction(); - action.setTaskPipe(TASK_PIPE_WITH_PREFIX); - assertTrue(PayloadBatchingService.hasSubdocumentBatcherPrefix(action)); - } - - @Test - public void testHasSubdocumentBatcherPrefix_WithoutPrefix() - { - final WorkerAction action = new WorkerAction(); - action.setTaskPipe(BASE_TASK_PIPE); - assertFalse(PayloadBatchingService.hasSubdocumentBatcherPrefix(action)); - } - - @Test - public void testHasSubdocumentBatcherPrefix_NullTaskPipe() - { - final WorkerAction action = new WorkerAction(); - action.setTaskPipe(null); - assertFalse(PayloadBatchingService.hasSubdocumentBatcherPrefix(action)); - } - - @Test - public void testStripBatcherPrefix_WithPrefix() - { - final WorkerAction action = new WorkerAction(); - action.setTaskPipe(TASK_PIPE_WITH_PREFIX); - assertEquals(BASE_TASK_PIPE, PayloadBatchingService.stripBatcherPrefix(action)); - } - - @Test - public void testStripBatcherPrefix_WithoutPrefix() - { - final WorkerAction action = new WorkerAction(); - action.setTaskPipe(BASE_TASK_PIPE); - assertEquals(BASE_TASK_PIPE, PayloadBatchingService.stripBatcherPrefix(action)); - } - - @Test - public void testStripBatcherPrefix_NullTaskPipe() - { - final WorkerAction action = new WorkerAction(); - action.setTaskPipe(null); - assertNull(PayloadBatchingService.stripBatcherPrefix(action)); - } - @Test public void testGetSubdocumentsCount_Valid() throws JsonProcessingException { diff --git a/job-service-util/src/main/java/com/github/jobservice/util/TaskPipeUtil.java b/job-service-util/src/main/java/com/github/jobservice/util/TaskPipeUtil.java new file mode 100644 index 000000000..71517c968 --- /dev/null +++ b/job-service-util/src/main/java/com/github/jobservice/util/TaskPipeUtil.java @@ -0,0 +1,64 @@ +/* + * Copyright 2016-2026 Open Text. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.jobservice.util; + +/** + * Utility class for task pipe handling. + *

+ * The "DocumentWorkerSubdocumentBatcher()" prefix is a directive that indicates + * a job should be evaluated for payload batching. This prefix should be stripped + * before sending messages to workers, as the actual target queue is the part + * after the prefix. + */ +public final class TaskPipeUtil +{ + + public static final String SUBDOCUMENT_BATCHER_PREFIX = "DocumentWorkerSubdocumentBatcher() "; + + private TaskPipeUtil() + { + } + + /** + * Checks if the task pipe starts with the DocumentWorkerSubdocumentBatcher() prefix. + * + * @param taskPipe The task pipe to check + * @return true if task pipe has the batching prefix, false otherwise + */ + public static boolean hasSubdocumentBatcherPrefix(final String taskPipe) + { + return taskPipe != null && taskPipe.startsWith(SUBDOCUMENT_BATCHER_PREFIX); + } + + /** + * Strips the DocumentWorkerSubdocumentBatcher() prefix from the task pipe if present. + *

+ * This method returns the actual target queue name that should be used for + * publishing messages. The prefix is only a directive for the job service, + * not part of the actual queue name. + * + * @param taskPipe The task pipe (may or may not have the prefix) + * @return The task pipe without the prefix, or the original task pipe if prefix not present + */ + public static String stripBatcherPrefix(final String taskPipe) + { + if (taskPipe != null && taskPipe.startsWith(SUBDOCUMENT_BATCHER_PREFIX)) { + return taskPipe.substring(SUBDOCUMENT_BATCHER_PREFIX.length()); + } + return taskPipe; + } +} + diff --git a/job-service-util/src/test/java/com/github/jobservice/util/TaskPipeUtilTest.java b/job-service-util/src/test/java/com/github/jobservice/util/TaskPipeUtilTest.java new file mode 100644 index 000000000..950b708a2 --- /dev/null +++ b/job-service-util/src/test/java/com/github/jobservice/util/TaskPipeUtilTest.java @@ -0,0 +1,125 @@ +/* + * Copyright 2016-2026 Open Text. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.github.jobservice.util; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Unit tests for TaskPipeUtil. + */ +public class TaskPipeUtilTest +{ + private static final String BASE_TASK_PIPE = "worker-queue"; + private static final String TASK_PIPE_WITH_PREFIX = TaskPipeUtil.SUBDOCUMENT_BATCHER_PREFIX + BASE_TASK_PIPE; + + // ===================================================== + // hasSubdocumentBatcherPrefix TESTS + // ===================================================== + + @Test + public void testHasSubdocumentBatcherPrefix_WithPrefix() + { + assertTrue(TaskPipeUtil.hasSubdocumentBatcherPrefix(TASK_PIPE_WITH_PREFIX)); + } + + @Test + public void testHasSubdocumentBatcherPrefix_WithoutPrefix() + { + assertFalse(TaskPipeUtil.hasSubdocumentBatcherPrefix(BASE_TASK_PIPE)); + } + + @Test + public void testHasSubdocumentBatcherPrefix_NullTaskPipe() + { + assertFalse(TaskPipeUtil.hasSubdocumentBatcherPrefix(null)); + } + + @Test + public void testHasSubdocumentBatcherPrefix_EmptyString() + { + assertFalse(TaskPipeUtil.hasSubdocumentBatcherPrefix("")); + } + + @Test + public void testHasSubdocumentBatcherPrefix_PartialPrefix() + { + // Should not match partial prefix + assertFalse(TaskPipeUtil.hasSubdocumentBatcherPrefix("DocumentWorkerSubdocument")); + } + + // ===================================================== + // stripBatcherPrefix TESTS + // ===================================================== + + @Test + public void testStripBatcherPrefix_WithPrefix() + { + assertEquals(BASE_TASK_PIPE, TaskPipeUtil.stripBatcherPrefix(TASK_PIPE_WITH_PREFIX)); + } + + @Test + public void testStripBatcherPrefix_WithoutPrefix() + { + assertEquals(BASE_TASK_PIPE, TaskPipeUtil.stripBatcherPrefix(BASE_TASK_PIPE)); + } + + @Test + public void testStripBatcherPrefix_NullTaskPipe() + { + assertNull(TaskPipeUtil.stripBatcherPrefix(null)); + } + + @Test + public void testStripBatcherPrefix_EmptyString() + { + assertEquals("", TaskPipeUtil.stripBatcherPrefix("")); + } + + @Test + public void testStripBatcherPrefix_OnlyPrefix() + { + // When task pipe is exactly the prefix, result should be empty + assertEquals("", TaskPipeUtil.stripBatcherPrefix(TaskPipeUtil.SUBDOCUMENT_BATCHER_PREFIX)); + } + + @Test + public void testStripBatcherPrefix_PrefixWithSpecialCharacters() + { + final String queueWithSpecialChars = "my-worker.queue/input"; + final String prefixed = TaskPipeUtil.SUBDOCUMENT_BATCHER_PREFIX + queueWithSpecialChars; + assertEquals(queueWithSpecialChars, TaskPipeUtil.stripBatcherPrefix(prefixed)); + } + + // ===================================================== + // CONSTANT TESTS + // ===================================================== + + @Test + public void testPrefixEndsWithSpace() + { + // The prefix should end with a space to separate from actual queue name + assertTrue(TaskPipeUtil.SUBDOCUMENT_BATCHER_PREFIX.endsWith(" ")); + } + + @Test + public void testPrefixValue() + { + assertEquals("DocumentWorkerSubdocumentBatcher() ", TaskPipeUtil.SUBDOCUMENT_BATCHER_PREFIX); + } +} + diff --git a/worker-jobtracking/src/main/java/com/github/jobservice/workers/jobtracking/JobTrackingWorkerUtil.java b/worker-jobtracking/src/main/java/com/github/jobservice/workers/jobtracking/JobTrackingWorkerUtil.java index d79c0e4f1..75de9d926 100644 --- a/worker-jobtracking/src/main/java/com/github/jobservice/workers/jobtracking/JobTrackingWorkerUtil.java +++ b/worker-jobtracking/src/main/java/com/github/jobservice/workers/jobtracking/JobTrackingWorkerUtil.java @@ -16,6 +16,7 @@ package com.github.jobservice.workers.jobtracking; import com.github.jobservice.util.JobTaskId; +import com.github.jobservice.util.TaskPipeUtil; import com.github.workerframework.api.TaskMessage; import com.github.workerframework.api.TaskStatus; import com.github.workerframework.api.TrackingInfo; @@ -63,6 +64,9 @@ public static TaskMessage createDependentJobTaskMessage(final JobTrackingWorkerD new JobTaskId(jobDependency.getPartitionId(), jobDependency.getJobId()).getMessageId(), new Date(), getStatusCheckIntervalMillis(statusCheckIntervalSeconds), statusCheckUrl, trackingPipe, jobDependency.getTargetPipe()); + // Strip the DocumentWorkerSubdocumentBatcher() prefix from taskPipe if present. + final String actualTaskPipe = TaskPipeUtil.stripBatcherPrefix(jobDependency.getTaskPipe()); + return new TaskMessage( taskId, jobDependency.getTaskClassifier(), @@ -70,7 +74,7 @@ public static TaskMessage createDependentJobTaskMessage(final JobTrackingWorkerD jobDependency.getTaskData(), TaskStatus.NEW_TASK, Collections.emptyMap(), - jobDependency.getTaskPipe(), + actualTaskPipe, trackingInfo, null, correlationId); From 7bb932f9f7899e37c273f6711277a26aab58825c Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Thu, 14 May 2026 12:16:06 +0530 Subject: [PATCH 3/9] US1124088: Fixing Test environment --- .../JobServicePayloadBatchingEndToEndIT.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java b/job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java index c1441e38a..f50367646 100644 --- a/job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java +++ b/job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java @@ -106,12 +106,17 @@ public static void setupClass() throws Exception // Create RabbitMQ connection for queue operations rabbitConn = RabbitUtil.createRabbitConnection(rabbitConfiguration.getRabbitConfiguration()); - final String connectionString = System.getenv("webserviceurl"); + jobsApi = createJobsApi(); + } + + private static JobsApi createJobsApi() + { + final String connectionString = System.getenv("CAF_WEBSERVICE_URL"); final ApiClient client = new ApiClient(); client.setBasePath(connectionString); final SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ"); client.setDateFormat(f); - jobsApi = new JobsApi(client); + return new JobsApi(client); } @BeforeMethod From be6662d727afef9a41301d64f9b49f5ba0d83812 Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Fri, 15 May 2026 14:53:42 +0530 Subject: [PATCH 4/9] US1124088: Dev build without prefix --- .../batching/PayloadBatchingService.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java index a8d8898c1..2e29a54b7 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java @@ -73,10 +73,10 @@ public static boolean shouldBatchPayload(final WorkerAction workerAction) return false; } - if (!TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { - LOG.debug("Task pipe does not have DocumentWorkerSubdocumentBatcher() prefix, skipping batching"); - return false; - } +// if (!TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { +// LOG.debug("Task pipe does not have DocumentWorkerSubdocumentBatcher() prefix, skipping batching"); +// return false; +// } final int subdocCount = getSubdocumentsCount(workerAction); if (subdocCount < 0) { From 079f5fa930102e4831d03042d21bbebf0ee2114a Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Fri, 15 May 2026 15:21:46 +0530 Subject: [PATCH 5/9] US1124088: Dev build without prefix --- .../JobServicePayloadBatchingIT.java | 50 +++++++++---------- .../batching/PayloadBatchingTest.java | 14 +++--- 2 files changed, 32 insertions(+), 32 deletions(-) diff --git a/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java b/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java index 28bdaaa7e..f744eb44d 100644 --- a/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java +++ b/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java @@ -323,31 +323,31 @@ public void testSmallDocumentWorkerTask_SingleMessage() throws Exception } // DocumentWorkerTask WITHOUT prefix but with large subdocs - single message (not opted in) - @Test - public void testDocumentWorkerTask_NoPrefix_SingleMessage() throws Exception - { - final String queueName = "payload-batching-noprefix-test-" + UUID.randomUUID(); - testQueueManager = getQueueManager(queueName); - - final int subdocCount = BATCH_SIZE * 3; // Above threshold but no prefix - final String jobId = UUID.randomUUID().toString(); - final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, false); // WITHOUT prefix - - // Start listening BEFORE creating the job - final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 1); - - // Create the job (messages will be published) - jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); - - // Wait for messages and get results (blocking) - final List received = messageSupplier.get(); - assertEquals(received.size(), 1, "Should receive single message when prefix not present (not opted in)"); - - // Verify no subtask suffix - final String taskId = received.get(0).getTaskId(); - assertFalse(SUBTASK_ID_PATTERN.matcher(taskId).matches(), - "DocumentWorkerTask without prefix should not have subtask suffix"); - } +// @Test +// public void testDocumentWorkerTask_NoPrefix_SingleMessage() throws Exception +// { +// final String queueName = "payload-batching-noprefix-test-" + UUID.randomUUID(); +// testQueueManager = getQueueManager(queueName); +// +// final int subdocCount = BATCH_SIZE * 3; // Above threshold but no prefix +// final String jobId = UUID.randomUUID().toString(); +// final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, false); // WITHOUT prefix +// +// // Start listening BEFORE creating the job +// final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 1); +// +// // Create the job (messages will be published) +// jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); +// +// // Wait for messages and get results (blocking) +// final List received = messageSupplier.get(); +// assertEquals(received.size(), 1, "Should receive single message when prefix not present (not opted in)"); +// +// // Verify no subtask suffix +// final String taskId = received.get(0).getTaskId(); +// assertFalse(SUBTASK_ID_PATTERN.matcher(taskId).matches(), +// "DocumentWorkerTask without prefix should not have subtask suffix"); +// } // Verify task pipe in batch messages has prefix stripped @Test diff --git a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java index d22761a1b..1404d0eb9 100644 --- a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java +++ b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java @@ -80,13 +80,13 @@ public void testDocumentWorkerTask_WithPrefix_LargeSubdocuments_SplittingEnabled assertTrue(PayloadBatchingService.shouldBatchPayload(action)); } - @Test - public void testDocumentWorkerTask_NoPrefix_LargeSubdocuments_NoSplitting() throws JsonProcessingException - { - // DocumentWorkerTask WITHOUT prefix but with large subdocuments (not opted in) - final WorkerAction action = createWorkerActionWithoutPrefix(createTaskDataMap(BATCH_SIZE * 5)); - assertFalse(PayloadBatchingService.shouldBatchPayload(action)); - } +// @Test +// public void testDocumentWorkerTask_NoPrefix_LargeSubdocuments_NoSplitting() throws JsonProcessingException +// { +// // DocumentWorkerTask WITHOUT prefix but with large subdocuments (not opted in) +// final WorkerAction action = createWorkerActionWithoutPrefix(createTaskDataMap(BATCH_SIZE * 5)); +// assertFalse(PayloadBatchingService.shouldBatchPayload(action)); +// } @Test public void testNonDocumentWorkerTask_WithPrefix_NoSplitting() throws JsonProcessingException From b2f6edfdf7ca66e055f4ccf0a17acb68f50f0669 Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Wed, 20 May 2026 14:57:14 +0530 Subject: [PATCH 6/9] US1124088: Uncommenting the prefix check --- .../JobServicePayloadBatchingIT.java | 50 +++++++++---------- .../batching/PayloadBatchingService.java | 8 +-- .../batching/PayloadBatchingTest.java | 14 +++--- 3 files changed, 36 insertions(+), 36 deletions(-) diff --git a/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java b/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java index f744eb44d..28bdaaa7e 100644 --- a/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java +++ b/job-service-container-tests/src/test/java/com/github/jobservice/containertests/JobServicePayloadBatchingIT.java @@ -323,31 +323,31 @@ public void testSmallDocumentWorkerTask_SingleMessage() throws Exception } // DocumentWorkerTask WITHOUT prefix but with large subdocs - single message (not opted in) -// @Test -// public void testDocumentWorkerTask_NoPrefix_SingleMessage() throws Exception -// { -// final String queueName = "payload-batching-noprefix-test-" + UUID.randomUUID(); -// testQueueManager = getQueueManager(queueName); -// -// final int subdocCount = BATCH_SIZE * 3; // Above threshold but no prefix -// final String jobId = UUID.randomUUID().toString(); -// final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, false); // WITHOUT prefix -// -// // Start listening BEFORE creating the job -// final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 1); -// -// // Create the job (messages will be published) -// jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); -// -// // Wait for messages and get results (blocking) -// final List received = messageSupplier.get(); -// assertEquals(received.size(), 1, "Should receive single message when prefix not present (not opted in)"); -// -// // Verify no subtask suffix -// final String taskId = received.get(0).getTaskId(); -// assertFalse(SUBTASK_ID_PATTERN.matcher(taskId).matches(), -// "DocumentWorkerTask without prefix should not have subtask suffix"); -// } + @Test + public void testDocumentWorkerTask_NoPrefix_SingleMessage() throws Exception + { + final String queueName = "payload-batching-noprefix-test-" + UUID.randomUUID(); + testQueueManager = getQueueManager(queueName); + + final int subdocCount = BATCH_SIZE * 3; // Above threshold but no prefix + final String jobId = UUID.randomUUID().toString(); + final NewJob newJob = createDocumentWorkerJob(jobId, subdocCount, queueName, false); // WITHOUT prefix + + // Start listening BEFORE creating the job + final Supplier> messageSupplier = startCollectingMessages(testQueueManager, 1); + + // Create the job (messages will be published) + jobsApi.createOrUpdateJob(defaultPartitionId, jobId, newJob, "1"); + + // Wait for messages and get results (blocking) + final List received = messageSupplier.get(); + assertEquals(received.size(), 1, "Should receive single message when prefix not present (not opted in)"); + + // Verify no subtask suffix + final String taskId = received.get(0).getTaskId(); + assertFalse(SUBTASK_ID_PATTERN.matcher(taskId).matches(), + "DocumentWorkerTask without prefix should not have subtask suffix"); + } // Verify task pipe in batch messages has prefix stripped @Test diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java index 2e29a54b7..a8d8898c1 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java @@ -73,10 +73,10 @@ public static boolean shouldBatchPayload(final WorkerAction workerAction) return false; } -// if (!TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { -// LOG.debug("Task pipe does not have DocumentWorkerSubdocumentBatcher() prefix, skipping batching"); -// return false; -// } + if (!TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { + LOG.debug("Task pipe does not have DocumentWorkerSubdocumentBatcher() prefix, skipping batching"); + return false; + } final int subdocCount = getSubdocumentsCount(workerAction); if (subdocCount < 0) { diff --git a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java index 1404d0eb9..d22761a1b 100644 --- a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java +++ b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java @@ -80,13 +80,13 @@ public void testDocumentWorkerTask_WithPrefix_LargeSubdocuments_SplittingEnabled assertTrue(PayloadBatchingService.shouldBatchPayload(action)); } -// @Test -// public void testDocumentWorkerTask_NoPrefix_LargeSubdocuments_NoSplitting() throws JsonProcessingException -// { -// // DocumentWorkerTask WITHOUT prefix but with large subdocuments (not opted in) -// final WorkerAction action = createWorkerActionWithoutPrefix(createTaskDataMap(BATCH_SIZE * 5)); -// assertFalse(PayloadBatchingService.shouldBatchPayload(action)); -// } + @Test + public void testDocumentWorkerTask_NoPrefix_LargeSubdocuments_NoSplitting() throws JsonProcessingException + { + // DocumentWorkerTask WITHOUT prefix but with large subdocuments (not opted in) + final WorkerAction action = createWorkerActionWithoutPrefix(createTaskDataMap(BATCH_SIZE * 5)); + assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + } @Test public void testNonDocumentWorkerTask_WithPrefix_NoSplitting() throws JsonProcessingException From efa7c5859af8f6e4b3a79f91abbaff664ca15ca7 Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Mon, 25 May 2026 11:25:59 +0530 Subject: [PATCH 7/9] US1124088: Updating Release notes. --- release-notes-10.1.2.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/release-notes-10.1.2.md b/release-notes-10.1.2.md index b36546391..48a9cf747 100644 --- a/release-notes-10.1.2.md +++ b/release-notes-10.1.2.md @@ -4,5 +4,8 @@ ${version-number} #### New Features +- US1124088: job-service enhancement. + - Splits the `DocumentWorkerTask` job with large payload containing many subdocuments into multiple smaller messages to prevent Out-of-Memory (OOM) errors while maintaining single job identity and existing progress tracking. + - Enabled via `DocumentWorkerSubdocumentBatcher() ` prefix in the task pipe of job definitions. #### Known Issues From 124f17a8a6dbe59e0e4ee69a9d10e2001391c3d3 Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Tue, 16 Jun 2026 12:09:03 +0530 Subject: [PATCH 8/9] US1124088: Code Review changes. --- .../scheduledexecutor/DatabasePoller.java | 3 + .../scheduledexecutor/QueueServices.java | 79 ++++++++++++------- .../ScheduledExecutorConfig.java | 8 ++ .../batching/PayloadBatchingService.java | 27 ++----- .../batching/SubdocumentBatchSplitter.java | 42 +++++++--- .../batching/PayloadBatchingTest.java | 72 ++++++++++++++--- release-notes-10.3.0.md | 3 +- 7 files changed, 164 insertions(+), 70 deletions(-) diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java index a9b808f0a..d9bf98b87 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java @@ -91,6 +91,9 @@ private static void sendMessageToQueueMessaging(final Codec codec, final JobTask // Strip the DocumentWorkerSubdocumentBatcher() prefix from taskPipe to get the actual target queue. final String actualTargetQueue = TaskPipeUtil.stripBatcherPrefix(workerAction.getTaskPipe()); + if( actualTargetQueue.isBlank()) { + throw new RuntimeException("Task pipe must not be empty after stripping batcher prefix."); + } try (final QueueServices queueServices = QueueServicesFactory.create(actualTargetQueue, partitionId, codec)) { diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java index 175cd708f..be18bb4c2 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java @@ -17,7 +17,9 @@ import com.github.cafapi.common.api.Codec; import com.github.cafapi.common.api.CodecException; +import com.github.jobservice.scheduledexecutor.batching.*; import com.github.jobservice.util.JobTaskId; +import com.github.jobservice.util.TaskPipeUtil; import com.github.workerframework.api.TaskMessage; import com.github.workerframework.api.TaskStatus; import com.github.workerframework.api.TrackingInfo; @@ -39,10 +41,7 @@ import java.util.UUID; import java.util.concurrent.TimeoutException; -import com.github.jobservice.scheduledexecutor.batching.PayloadBatchingService; -import com.github.jobservice.scheduledexecutor.batching.SubdocumentBatchSplitter; -import com.github.jobservice.scheduledexecutor.batching.SubtaskIdGenerator; -import com.github.jobservice.util.TaskPipeUtil; +import static com.github.jobservice.scheduledexecutor.batching.PayloadBatchingService.DOCUMENT_WORKER_TASK_CLASSIFIER; /** * This class is responsible for sending task data to the target queue. @@ -83,9 +82,16 @@ public void sendMessage( final String partitionId, final String jobId, final WorkerAction workerAction ) throws IOException, URISyntaxException, InterruptedException, TimeoutException { + // If not opted for payload batching, send the message as a single task message. + if (!TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { + sendSingleMessage(partitionId, jobId, workerAction); + return; + } + // Check if payload batching is required - if (PayloadBatchingService.shouldBatchPayload(workerAction)) { - sendBatchedMessages(partitionId, jobId, workerAction); + Map taskDataMap = PayloadBatchingService.deserializeTaskData(workerAction); + if (PayloadBatchingService.shouldBatchPayload(workerAction, taskDataMap)) { + sendBatchedMessages(partitionId, jobId, workerAction, taskDataMap); } else { sendSingleMessage(partitionId, jobId, workerAction); } @@ -117,22 +123,20 @@ private void sendSingleMessage( statusCheckUrl, ScheduledExecutorConfig.getTrackingPipe(), workerAction.getTargetPipe()); // Strip the batcher prefix from task pipe if present - final WorkerAction NonBatchingWorkerAction; + WorkerAction nonBatchingWorkerAction; if (TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { final String strippedTaskPipe = TaskPipeUtil.stripBatcherPrefix(workerAction.getTaskPipe()); - NonBatchingWorkerAction = new WorkerAction(); - NonBatchingWorkerAction.setTaskClassifier(workerAction.getTaskClassifier()); - NonBatchingWorkerAction.setTaskApiVersion(workerAction.getTaskApiVersion()); - NonBatchingWorkerAction.setTaskData(workerAction.getTaskData()); - NonBatchingWorkerAction.setTaskDataEncoding(workerAction.getTaskDataEncoding()); - NonBatchingWorkerAction.setTaskPipe(strippedTaskPipe); - NonBatchingWorkerAction.setTargetPipe(workerAction.getTargetPipe()); - NonBatchingWorkerAction.setCorrelationId(workerAction.getCorrelationId()); + if( strippedTaskPipe.isBlank()) { + throw new RuntimeException("Task pipe must not be empty after stripping batcher prefix."); + } + nonBatchingWorkerAction = new WorkerAction(); + nonBatchingWorkerAction = CopyWorkerActionWithStrippedTaskpipe(workerAction, nonBatchingWorkerAction, strippedTaskPipe); + } else { - NonBatchingWorkerAction = workerAction; + nonBatchingWorkerAction = workerAction; } - final Object taskMessage = getTaskMessage(trackingInfo, NonBatchingWorkerAction, taskId); + final Object taskMessage = getTaskMessage(trackingInfo, nonBatchingWorkerAction, taskId); // Serialise and publish publishTaskMessage(taskMessage); @@ -144,11 +148,10 @@ private void sendSingleMessage( private void sendBatchedMessages( final String partitionId, final String jobId, - final WorkerAction workerAction + final WorkerAction workerAction, + final Map taskDataMap ) throws IOException, URISyntaxException, InterruptedException, TimeoutException { - // Deserialize taskData JSON to Map - final Map taskDataMap = PayloadBatchingService.deserializeTaskData(workerAction); if (taskDataMap == null) { throw new RuntimeException("Failed to deserialize taskData for batching"); } @@ -175,19 +178,27 @@ private void sendBatchedMessages( // Strip the DocumentWorkerSubdocumentBatcher() prefix from task pipe final String strippedTaskPipe = TaskPipeUtil.stripBatcherPrefix(workerAction.getTaskPipe()); + if( strippedTaskPipe.isBlank()) { + throw new RuntimeException("Task pipe must not be empty after stripping batcher prefix."); + } + LOG.info("Sending {} batched messages for job {} ({} subdocuments, batch size {}, task pipe: {})", totalBatches, jobId, subdocuments.size(), batchSize, strippedTaskPipe); + // Build a reusable base map once so each batch only injects subdocuments. + final Map taskDataTemplateWithoutSubdocs = + SubdocumentBatchSplitter.createTaskDataTemplateWithoutSubdocuments(taskDataMap); + // Process ONE batch at a time for (int batchIndex = 1; batchIndex <= totalBatches; batchIndex++) { - // 1. Get subdocuments subList VIEW for this batch + // 1. Get subdocuments subList VIEW for this batch. Must not be used after the underlying taskData map is mutated to avoid ConcurrentModificationException. final List subdocBatch = SubdocumentBatchSplitter.getSubdocumentsBatchView( subdocuments, batchIndex, batchSize); // 2. Create batched taskData Map with this batch's subdocuments - final Map batchedTaskDataMap = SubdocumentBatchSplitter.createBatchedTaskData( - taskDataMap, subdocBatch); + final Map batchedTaskDataMap = SubdocumentBatchSplitter.createBatchedTaskDataFromTemplate( + taskDataTemplateWithoutSubdocs, subdocBatch); // 3. Serialize batched taskData to JSON String for WorkerAction final byte[] batchedTaskDataBytes; @@ -214,14 +225,9 @@ private void sendBatchedMessages( workerAction.getTargetPipe()); // 6. Create a modified WorkerAction with batched taskData and stripped task pipe - final WorkerAction batchWorkerAction = new WorkerAction(); - batchWorkerAction.setTaskClassifier(workerAction.getTaskClassifier()); - batchWorkerAction.setTaskApiVersion(workerAction.getTaskApiVersion()); + WorkerAction batchWorkerAction = new WorkerAction(); + batchWorkerAction = CopyWorkerActionWithStrippedTaskpipe(workerAction, batchWorkerAction, strippedTaskPipe); batchWorkerAction.setTaskData(batchedTaskDataString); - batchWorkerAction.setTaskDataEncoding(WorkerAction.TaskDataEncodingEnum.UTF8); - batchWorkerAction.setTaskPipe(strippedTaskPipe); - batchWorkerAction.setTargetPipe(workerAction.getTargetPipe()); - batchWorkerAction.setCorrelationId(workerAction.getCorrelationId()); // 7. Reuse existing getTaskMessage() method final TaskMessage taskMessage = getTaskMessage(trackingInfo, batchWorkerAction, taskSubtaskId); @@ -346,4 +352,17 @@ public void close() throws RuntimeException{ } } + private static WorkerAction CopyWorkerActionWithStrippedTaskpipe(WorkerAction orgWorkerAction, WorkerAction copyWorkerAction, String strippedTaskPipe) + { + copyWorkerAction.setTaskClassifier(orgWorkerAction.getTaskClassifier()); + copyWorkerAction.setTaskApiVersion(orgWorkerAction.getTaskApiVersion()); + copyWorkerAction.setTaskData(orgWorkerAction.getTaskData()); + copyWorkerAction.setTaskDataEncoding(orgWorkerAction.getTaskDataEncoding()); + copyWorkerAction.setTaskPipe(strippedTaskPipe); + copyWorkerAction.setTargetPipe(orgWorkerAction.getTargetPipe()); + copyWorkerAction.setCorrelationId(orgWorkerAction.getCorrelationId()); + + return copyWorkerAction; + } + } diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/ScheduledExecutorConfig.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/ScheduledExecutorConfig.java index 2479bee7e..b62a1604c 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/ScheduledExecutorConfig.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/ScheduledExecutorConfig.java @@ -189,4 +189,12 @@ public static int getJobProgressUpdatePeriod() { } return Integer.parseInt(period); } + + public static int getJobServicePayloadBatchingSize() { + final String batchSize = getPropertyOrEnvVar("JOB_SERVICE_PAYLOAD_BATCH_SIZE"); + if( null == batchSize || batchSize.isEmpty()) { + return 200; + } + return Integer.parseInt(batchSize); + } } diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java index a8d8898c1..447906322 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java @@ -18,6 +18,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.jobservice.scheduledexecutor.ScheduledExecutorConfig; import com.github.jobservice.scheduledexecutor.WorkerAction; import com.github.jobservice.util.TaskPipeUtil; import org.slf4j.Logger; @@ -46,7 +47,7 @@ public final class PayloadBatchingService /** Maximum number of subdocuments per batch */ - public static final int BATCH_SIZE = 200; + public static final int BATCH_SIZE = ScheduledExecutorConfig.getJobServicePayloadBatchingSize(); private PayloadBatchingService() { @@ -64,9 +65,10 @@ private PayloadBatchingService() * * * @param workerAction The worker action to evaluate + * @param taskDataMap to extract subdocuments count to evaluate * @return true if the payload should be batched, false otherwise */ - public static boolean shouldBatchPayload(final WorkerAction workerAction) + public static boolean shouldBatchPayload(final WorkerAction workerAction, Map taskDataMap) { if (!DOCUMENT_WORKER_TASK_CLASSIFIER.equals(workerAction.getTaskClassifier())) { LOG.debug("Not a DocumentWorkerTask, skipping batching"); @@ -78,7 +80,7 @@ public static boolean shouldBatchPayload(final WorkerAction workerAction) return false; } - final int subdocCount = getSubdocumentsCount(workerAction); + final int subdocCount = getSubdocumentsCount(taskDataMap); if (subdocCount < 0) { LOG.debug("Could not determine subdocuments count, skipping batching"); return false; @@ -98,12 +100,11 @@ public static boolean shouldBatchPayload(final WorkerAction workerAction) /** * Gets the subdocuments count from a worker action. - * @param workerAction The worker action + * @param taskDataMap To extract subdocuments count * @return Subdocuments count, or -1 if cannot be determined */ - public static int getSubdocumentsCount(final WorkerAction workerAction) + public static int getSubdocumentsCount(final Map taskDataMap) { - final Map taskDataMap = deserializeTaskData(workerAction); if (taskDataMap == null) { return -1; } @@ -143,20 +144,6 @@ public static Map deserializeTaskData(final WorkerAction workerA } } - /** - * Calculates the total number of batches needed. - * @param workerAction The worker action - * @return Number of batches, or 0 if batching not applicable - */ - public static int calculateTotalBatches(final WorkerAction workerAction) - { - final int subdocCount = getSubdocumentsCount(workerAction); - if (subdocCount <= 0) { - return 0; - } - return SubdocumentBatchSplitter.calculateBatchCount(subdocCount, BATCH_SIZE); - } - /** * Gets the batch size constant. * @return The batch size (200) diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java index d72e0cf9e..2b9b3e46c 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java @@ -79,7 +79,7 @@ public static int calculateBatchCount(final int totalItems, final int batchSize) if (batchSize < 1) { throw new IllegalArgumentException("Batch size must be >= 1"); } - return (int) Math.ceil((double) totalItems / batchSize); + return Math.ceilDiv(totalItems,batchSize); } /** @@ -127,21 +127,45 @@ public static List getSubdocumentsBatchView( } /** - * Creates a new taskData map with a batch of subdocuments replacing the original. + * Creates a reusable taskData template with document.subdocuments removed. + * This avoids re-copying unchanged top-level and document entries for every batch. + * * @param originalTaskData The original taskData map - * @param subdocumentsBatch The batch of subdocuments to use (can be a subList view) - * @return New taskData map with batched subdocuments + * @return A new template map without document.subdocuments */ @SuppressWarnings("unchecked") - public static Map createBatchedTaskData( - final Map originalTaskData, - final List subdocumentsBatch) + public static Map createTaskDataTemplateWithoutSubdocuments( + final Map originalTaskData) { - final Map batchedTaskData = new HashMap<>(originalTaskData); + final Map taskDataTemplate = new HashMap<>(originalTaskData); final Map originalDocument = (Map) originalTaskData.get(DOCUMENT_KEY); if (originalDocument != null) { - final Map batchedDocument = new HashMap<>(originalDocument); + final Map documentTemplate = new HashMap<>(originalDocument); + documentTemplate.remove(SUBDOCUMENTS_KEY); + taskDataTemplate.put(DOCUMENT_KEY, documentTemplate); + } + + return taskDataTemplate; + } + + /** + * Creates a new taskData map from a reusable template and batch subdocuments. + * + * @param taskDataTemplate A taskData template created without document.subdocuments + * @param subdocumentsBatch The batch of subdocuments to use (can be a subList view) + * @return New taskData map with batch subdocuments added into document + */ + @SuppressWarnings("unchecked") + public static Map createBatchedTaskDataFromTemplate( + final Map taskDataTemplate, + final List subdocumentsBatch) + { + final Map batchedTaskData = new HashMap<>(taskDataTemplate); + final Map documentTemplate = (Map) taskDataTemplate.get(DOCUMENT_KEY); + + if (documentTemplate != null) { + final Map batchedDocument = new HashMap<>(documentTemplate); batchedDocument.put(SUBDOCUMENTS_KEY, subdocumentsBatch); batchedTaskData.put(DOCUMENT_KEY, batchedDocument); } diff --git a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java index d22761a1b..cf7f6bb6e 100644 --- a/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java +++ b/job-service-scheduled-executor/src/test/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingTest.java @@ -53,7 +53,8 @@ public void testDocumentWorkerTask_WithPrefix_NoSubdocuments_NoSplitting() throw { // DocumentWorkerTask with prefix but without subdocuments field final WorkerAction action = createWorkerActionWithPrefix(createTaskDataWithoutSubdocs()); - assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); } @Test @@ -61,7 +62,8 @@ public void testDocumentWorkerTask_WithPrefix_EmptySubdocuments_NoSplitting() th { // DocumentWorkerTask with prefix and empty subdocuments array final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(0)); - assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); } @Test @@ -69,7 +71,8 @@ public void testDocumentWorkerTask_WithPrefix_SmallSubdocuments_NoSplitting() th { // DocumentWorkerTask with prefix and subdocuments below threshold final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(BATCH_SIZE / 2)); - assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); } @Test @@ -77,7 +80,8 @@ public void testDocumentWorkerTask_WithPrefix_LargeSubdocuments_SplittingEnabled { // DocumentWorkerTask with prefix and subdocuments above threshold final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(BATCH_SIZE + 1)); - assertTrue(PayloadBatchingService.shouldBatchPayload(action)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertTrue(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); } @Test @@ -85,7 +89,8 @@ public void testDocumentWorkerTask_NoPrefix_LargeSubdocuments_NoSplitting() thro { // DocumentWorkerTask WITHOUT prefix but with large subdocuments (not opted in) final WorkerAction action = createWorkerActionWithoutPrefix(createTaskDataMap(BATCH_SIZE * 5)); - assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); } @Test @@ -94,7 +99,8 @@ public void testNonDocumentWorkerTask_WithPrefix_NoSplitting() throws JsonProces // Non-DocumentWorkerTask with prefix in task pipe (wrong classifier) final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(BATCH_SIZE * 5)); action.setTaskClassifier("NonDocumentWorkerTask"); - assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); } @Test @@ -104,21 +110,25 @@ public void testShouldBatch_NullTaskData_ReturnsFalse() action.setTaskClassifier(PayloadBatchingService.DOCUMENT_WORKER_TASK_CLASSIFIER); action.setTaskPipe(TASK_PIPE_WITH_PREFIX); action.setTaskData(null); - assertFalse(PayloadBatchingService.shouldBatchPayload(action)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); } @Test public void testGetSubdocumentsCount_Valid() throws JsonProcessingException { final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(500)); - assertEquals(500, PayloadBatchingService.getSubdocumentsCount(action)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertEquals(500, PayloadBatchingService.getSubdocumentsCount(taskDataMap)); } @Test public void testCalculateTotalBatches() throws JsonProcessingException { final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(500)); - assertEquals(3, PayloadBatchingService.calculateTotalBatches(action)); // 500/200 = 2.5 = 3 + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + final int subdocCount = PayloadBatchingService.getSubdocumentsCount(taskDataMap); + assertEquals(Math.ceilDiv(500,BATCH_SIZE), SubdocumentBatchSplitter.calculateBatchCount(subdocCount, BATCH_SIZE)); // 500/200 = 2.5 = 3 } @Test @@ -243,10 +253,13 @@ public void testCreateBatchedTaskData_PreservesOtherFields() { final Map original = createTaskDataMap(500); final List subdocs = SubdocumentBatchSplitter.extractSubdocuments(original); + final Map template = + SubdocumentBatchSplitter.createTaskDataTemplateWithoutSubdocuments(original); for (int i = 1; i <= 3; i++) { final List batch = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, i, BATCH_SIZE); - final Map batched = SubdocumentBatchSplitter.createBatchedTaskData(original, batch); + final Map batched = + SubdocumentBatchSplitter.createBatchedTaskDataFromTemplate(template, batch); // Verify customData and scripts preserved assertEquals(original.get("customData"), batched.get("customData")); @@ -283,6 +296,45 @@ public void testCreateBatchedTaskData_CorrectSubdocumentsInEachBatch() assertEquals("500", getSubdocId(batch3.get(batch3.size() - 1))); } + @Test + @SuppressWarnings("unchecked") + public void testCreateTaskDataTemplateWithoutSubdocuments_RemovesOnlySubdocuments() + { + final Map original = createTaskDataMap(500); + + final Map template = + SubdocumentBatchSplitter.createTaskDataTemplateWithoutSubdocuments(original); + + final Map originalDoc = (Map) original.get("document"); + final Map templateDoc = (Map) template.get("document"); + + assertNotNull(templateDoc.get("fields")); + assertFalse(templateDoc.containsKey("subdocuments")); + assertTrue(originalDoc.containsKey("subdocuments")); + assertEquals(original.get("customData"), template.get("customData")); + assertEquals(original.get("scripts"), template.get("scripts")); + } + + @Test + @SuppressWarnings("unchecked") + public void testCreateBatchedTaskDataFromTemplate_InsertsBatchSubdocuments() + { + final Map original = createTaskDataMap(500); + final List subdocs = SubdocumentBatchSplitter.extractSubdocuments(original); + final Map template = + SubdocumentBatchSplitter.createTaskDataTemplateWithoutSubdocuments(original); + final List batch = SubdocumentBatchSplitter.getSubdocumentsBatchView(subdocs, 2, BATCH_SIZE); + + final Map batched = + SubdocumentBatchSplitter.createBatchedTaskDataFromTemplate(template, batch); + final Map batchedDoc = (Map) batched.get("document"); + final Map templateDoc = (Map) template.get("document"); + + assertEquals(BATCH_SIZE, ((List) batchedDoc.get("subdocuments")).size()); + assertEquals("201", getSubdocId(((List) batchedDoc.get("subdocuments")).get(0))); + assertFalse(templateDoc.containsKey("subdocuments")); + } + @Test public void testBatching_NoSubdocumentDuplication() { diff --git a/release-notes-10.3.0.md b/release-notes-10.3.0.md index 48a9cf747..e415d07a7 100644 --- a/release-notes-10.3.0.md +++ b/release-notes-10.3.0.md @@ -6,6 +6,7 @@ ${version-number} #### New Features - US1124088: job-service enhancement. - Splits the `DocumentWorkerTask` job with large payload containing many subdocuments into multiple smaller messages to prevent Out-of-Memory (OOM) errors while maintaining single job identity and existing progress tracking. - - Enabled via `DocumentWorkerSubdocumentBatcher() ` prefix in the task pipe of job definitions. + - Enabled via `DocumentWorkerSubdocumentBatcher() ` prefix in the task pipe for top-level job definitions. + - Added environment variable `JOB_SERVICE_PAYLOAD_BATCH_SIZE` in job-service-scheduled-executor config to configure the batch size which defaults to `200` if not found. #### Known Issues From 9c52ae5ad1d606bc6cf7aed22ca7df171a8a7a74 Mon Sep 17 00:00:00 2001 From: Rajasekar Balasubramani Date: Tue, 16 Jun 2026 12:52:37 +0530 Subject: [PATCH 9/9] US1124088: Adding a manual retry mechanism for sending batched messages --- .../scheduledexecutor/QueueServices.java | 38 ++++++++++++++++++- 1 file changed, 37 insertions(+), 1 deletion(-) diff --git a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java index be18bb4c2..0ebdbc903 100644 --- a/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/QueueServices.java @@ -39,6 +39,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import static com.github.jobservice.scheduledexecutor.batching.PayloadBatchingService.DOCUMENT_WORKER_TASK_CLASSIFIER; @@ -52,6 +53,8 @@ public final class QueueServices implements AutoCloseable private static final int RABBIT_MQ_PUBLISH_TIMEOUT_MILLIS = ScheduledExecutorConfig.getRabbitMQPublishTimeoutSeconds() * 1000; + private static final int BATCH_MESSAGE_MAX_RETRIES = 3; + private static final long BATCH_MESSAGE_RETRY_DELAY_MILLIS = 500L; private final Connection connection; private final Channel publisherChannel; @@ -236,12 +239,45 @@ private void sendBatchedMessages( LOG.debug("Sending batch {}/{} with {} subdocuments (taskId={}, jobTaskId={})", batchIndex, totalBatches, subdocBatch.size(), taskSubtaskId, jobTaskSubtaskId); - publishTaskMessage(taskMessage); + publishBatchedTaskMessageWithRetry(taskMessage, jobId, batchIndex, totalBatches); } LOG.info("Successfully sent all {} batches for job {}", totalBatches, jobId); } + /** + * Retries transient publish failures for individual batches to avoid failing the whole batched send. + */ + private void publishBatchedTaskMessageWithRetry( + final TaskMessage taskMessage, + final String jobId, + final int batchIndex, + final int totalBatches + ) throws IOException, InterruptedException, TimeoutException + { + int retryCount = 0; + + for (;;) { + try { + publishTaskMessage(taskMessage); + return; + } catch (final IOException | TimeoutException ex) { + if (retryCount >= BATCH_MESSAGE_MAX_RETRIES) { + LOG.error("Exceeded transient retry limit for job {} batch {}/{} after {} retries", + jobId, batchIndex, totalBatches, retryCount, ex); + throw ex; + } + + retryCount++; + LOG.warn("Transient failure sending job {} batch {}/{} (retry {}/{}). Retrying in {} ms", + jobId, batchIndex, totalBatches, retryCount, BATCH_MESSAGE_MAX_RETRIES, + BATCH_MESSAGE_RETRY_DELAY_MILLIS, ex); + + TimeUnit.MILLISECONDS.sleep(BATCH_MESSAGE_RETRY_DELAY_MILLIS); + } + } + } + /** * Publishes a task message to RabbitMQ. */