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..f50367646 --- /dev/null +++ b/job-service-acceptance-tests/src/test/java/com/github/jobservice/acceptancetests/JobServicePayloadBatchingEndToEndIT.java @@ -0,0 +1,455 @@ +/* + * 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()); + + 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); + return 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 3ff97a5dc..ae3ae878c 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/DatabasePoller.java b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/DatabasePoller.java index 680c2b8ee..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 @@ -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,13 @@ 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()); + 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)) { 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 6aeff23c0..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 @@ -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; @@ -34,10 +36,14 @@ 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.TimeUnit; import java.util.concurrent.TimeoutException; +import static com.github.jobservice.scheduledexecutor.batching.PayloadBatchingService.DOCUMENT_WORKER_TASK_CLASSIFIER; + /** * This class is responsible for sending task data to the target queue. */ @@ -47,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; @@ -76,6 +84,28 @@ public QueueServices( 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 + Map taskDataMap = PayloadBatchingService.deserializeTaskData(workerAction); + if (PayloadBatchingService.shouldBatchPayload(workerAction, taskDataMap)) { + sendBatchedMessages(partitionId, jobId, workerAction, taskDataMap); + } 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 +125,165 @@ 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 + WorkerAction nonBatchingWorkerAction; + if (TaskPipeUtil.hasSubdocumentBatcherPrefix(workerAction.getTaskPipe())) { + final String strippedTaskPipe = TaskPipeUtil.stripBatcherPrefix(workerAction.getTaskPipe()); + 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; + } + + 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, + final Map taskDataMap + ) throws IOException, URISyntaxException, InterruptedException, TimeoutException + { + 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 = 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); - // Serialise the task message. - // Wrap any CodecException as a RuntimeException as it shouldn't happen + // 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. 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.createBatchedTaskDataFromTemplate( + taskDataTemplateWithoutSubdocs, 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); + + // 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 + WorkerAction batchWorkerAction = new WorkerAction(); + batchWorkerAction = CopyWorkerActionWithStrippedTaskpipe(workerAction, batchWorkerAction, strippedTaskPipe); + batchWorkerAction.setTaskData(batchedTaskDataString); + + // 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); + + 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. + */ + private void publishTaskMessage(final Object taskMessage) + throws IOException, InterruptedException, TimeoutException + { final byte[] taskMessageBytes; try { LOG.debug("Serialise the task message ..."); @@ -203,4 +388,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 new file mode 100644 index 000000000..447906322 --- /dev/null +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/PayloadBatchingService.java @@ -0,0 +1,156 @@ +/* + * 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.ScheduledExecutorConfig; +import com.github.jobservice.scheduledexecutor.WorkerAction; +import com.github.jobservice.util.TaskPipeUtil; +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"; + + + /** Maximum number of subdocuments per batch */ + public static final int BATCH_SIZE = ScheduledExecutorConfig.getJobServicePayloadBatchingSize(); + + 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 + * @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, Map taskDataMap) + { + if (!DOCUMENT_WORKER_TASK_CLASSIFIER.equals(workerAction.getTaskClassifier())) { + LOG.debug("Not a DocumentWorkerTask, 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(taskDataMap); + 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; + } + + + /** + * Gets the subdocuments count from a worker action. + * @param taskDataMap To extract subdocuments count + * @return Subdocuments count, or -1 if cannot be determined + */ + public static int getSubdocumentsCount(final Map taskDataMap) + { + 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; + } + } + + /** + * 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..2b9b3e46c --- /dev/null +++ b/job-service-scheduled-executor/src/main/java/com/github/jobservice/scheduledexecutor/batching/SubdocumentBatchSplitter.java @@ -0,0 +1,176 @@ +/* + * 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 Math.ceilDiv(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 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 + * @return A new template map without document.subdocuments + */ + @SuppressWarnings("unchecked") + public static Map createTaskDataTemplateWithoutSubdocuments( + final Map originalTaskData) + { + final Map taskDataTemplate = new HashMap<>(originalTaskData); + final Map originalDocument = (Map) originalTaskData.get(DOCUMENT_KEY); + + if (originalDocument != null) { + 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); + } + + 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..cf7f6bb6e --- /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 com.github.jobservice.util.TaskPipeUtil; +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 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 = TaskPipeUtil.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()); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); + } + + @Test + public void testDocumentWorkerTask_WithPrefix_EmptySubdocuments_NoSplitting() throws JsonProcessingException + { + // DocumentWorkerTask with prefix and empty subdocuments array + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(0)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); + } + + @Test + public void testDocumentWorkerTask_WithPrefix_SmallSubdocuments_NoSplitting() throws JsonProcessingException + { + // DocumentWorkerTask with prefix and subdocuments below threshold + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(BATCH_SIZE / 2)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); + } + + @Test + public void testDocumentWorkerTask_WithPrefix_LargeSubdocuments_SplittingEnabled() throws JsonProcessingException + { + // DocumentWorkerTask with prefix and subdocuments above threshold + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(BATCH_SIZE + 1)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertTrue(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); + } + + @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)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); + } + + @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"); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); + } + + @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); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertFalse(PayloadBatchingService.shouldBatchPayload(action, taskDataMap)); + } + + @Test + public void testGetSubdocumentsCount_Valid() throws JsonProcessingException + { + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(500)); + final Map taskDataMap = PayloadBatchingService.deserializeTaskData(action); + assertEquals(500, PayloadBatchingService.getSubdocumentsCount(taskDataMap)); + } + + @Test + public void testCalculateTotalBatches() throws JsonProcessingException + { + final WorkerAction action = createWorkerActionWithPrefix(createTaskDataMap(500)); + 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 + 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); + 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.createBatchedTaskDataFromTemplate(template, 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 + @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() + { + 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"); + } +} 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/release-notes-10.3.0.md b/release-notes-10.3.0.md index b36546391..e415d07a7 100644 --- a/release-notes-10.3.0.md +++ b/release-notes-10.3.0.md @@ -4,5 +4,9 @@ ${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 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 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);