From e108a5e112379745624c09e619892f2ca05015ef Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Wed, 17 Jun 2026 14:55:11 -0300 Subject: [PATCH 1/4] Dead-letter Kafka records individually within a batch The consumer now reads records in batches (Message). Batch ack/nack apply to the whole batch, so a single bad record cannot be nacked on its own without affecting the rest. Instead of nacking, each failed record is produced to the data-index-events-dlq topic via a dedicated emitter, and the batch is acked exactly once after all dead-letter writes complete. The batch offset is committed only after every dead-letter write has finished, so a failed record is never dropped: if a dead-letter write itself fails the batch is not committed (failure-strategy=fail) and the records are reprocessed. Successfully processed records remain committed via the same ack. Replaces the incoming dead-letter-queue failure strategy with an outgoing data-index-events-dlq channel. Signed-off-by: Matheus Cruz --- .../kafka/service/KafkaLifecycleConsumer.java | 90 ++++++++++++++----- .../src/main/resources/application.properties | 14 ++- 2 files changed, 79 insertions(+), 25 deletions(-) diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java index 5ad0d66e5..e15eaa43c 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java @@ -21,10 +21,17 @@ import io.cloudevents.jackson.JsonCloudEventData; import io.serverlessworkflow.impl.lifecycle.ce.TaskCEData; import io.serverlessworkflow.impl.lifecycle.ce.WorkflowCEData; +import io.smallrye.reactive.messaging.MutinyEmitter; +import io.smallrye.reactive.messaging.kafka.api.OutgoingKafkaRecordMetadata; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.eclipse.microprofile.reactive.messaging.Channel; import org.eclipse.microprofile.reactive.messaging.Incoming; +import org.eclipse.microprofile.reactive.messaging.Message; +import org.eclipse.microprofile.reactive.messaging.Metadata; import org.kubesmarts.logic.dataindex.ingestion.kafka.processor.EventProcessor; import org.kubesmarts.logic.dataindex.ingestion.kafka.processor.ProcessEventFailedException; import org.kubesmarts.logic.dataindex.model.LifecycleEventUtils; @@ -33,6 +40,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + @ApplicationScoped public class KafkaLifecycleConsumer { @@ -47,34 +60,69 @@ public class KafkaLifecycleConsumer { @Inject EventProcessor taskExecutionProcessor; + @Inject + @Channel("data-index-events-dlq") + MutinyEmitter deadLetterEmitter; + @Incoming("data-index-events") - public void consumeLifecycleEvent(ConsumerRecord record) { + public CompletionStage consumeLifecycleEvent(Message> records) { + List> deadLetterSends = new ArrayList<>(); - try { - CloudEvent cloudEvent = validateCloudEvent(record); + for (ConsumerRecord record : records.getPayload()) { + try { + CloudEvent cloudEvent = validateCloudEvent(record); - JsonCloudEventData cloudEventData = (JsonCloudEventData) cloudEvent.getData(); - if (cloudEventData == null || cloudEventData.getNode() == null) { - throw new IllegalArgumentException("The CloudEvent data node consumed at offset %s from partition %s is null or empty." - .formatted(record.offset(), record.partition())); - } + JsonCloudEventData cloudEventData = (JsonCloudEventData) cloudEvent.getData(); + if (cloudEventData == null || cloudEventData.getNode() == null) { + throw new IllegalArgumentException("The CloudEvent data node consumed at offset %s from partition %s is null or empty." + .formatted(record.offset(), record.partition())); + } - Class eventClass = LifecycleEventUtils.getEventClass(cloudEvent.getType()); - Object data = jackson.convertValue(cloudEventData.getNode(), eventClass); + Class eventClass = LifecycleEventUtils.getEventClass(cloudEvent.getType()); + Object data = jackson.convertValue(cloudEventData.getNode(), eventClass); - if (data instanceof TaskCEData taskData) { - handleTaskEvent(cloudEvent, taskData); - } else if (data instanceof WorkflowCEData workflowData) { - handleWorkflowEvent(cloudEvent, workflowData); - } else { - throw new IllegalArgumentException("Unsupported event type '%s' consumed at offset %s from partition %s." - .formatted(cloudEvent.getType(), record.offset(), record.partition())); + if (data instanceof TaskCEData taskData) { + handleTaskEvent(cloudEvent, taskData); + } else if (data instanceof WorkflowCEData workflowData) { + handleWorkflowEvent(cloudEvent, workflowData); + } else { + throw new IllegalArgumentException("Unsupported event type '%s' consumed at offset %s from partition %s." + .formatted(cloudEvent.getType(), record.offset(), record.partition())); + } + } catch (Exception e) { + log.error("Failed to consume the record from Kafka at offset '{}' from partition '{}'. Routing to dead-letter queue.", + record.offset(), record.partition(), e); + deadLetterSends.add(sendToDeadLetterQueue(record, e)); } - } catch (Exception e) { - log.error("Failed to consume the record from Kafka at offset '{}' from partition '{}'.", record.offset(), record.partition(), e); - throw new ProcessEventFailedException("Failed to consume Kafka record at offset %s from partition %s".formatted( - record.offset(), record.partition()), e); } + + // Commit the batch only after every dead-letter write has completed + CompletableFuture[] pending = deadLetterSends.stream() + .map(CompletionStage::toCompletableFuture) + .toArray(CompletableFuture[]::new); + + return CompletableFuture.allOf(pending).thenCompose(ignored -> records.ack()); + } + + private CompletionStage sendToDeadLetterQueue(ConsumerRecord record, Exception cause) { + RecordHeaders headers = new RecordHeaders(); + headers.add("dead-letter-reason", bytes(cause.getMessage() != null ? cause.getMessage() : cause.toString())); + headers.add("dead-letter-cause", bytes(cause.getClass().getName())); + headers.add("dead-letter-original-topic", bytes(record.topic())); + headers.add("dead-letter-original-partition", bytes(Integer.toString(record.partition()))); + headers.add("dead-letter-original-offset", bytes(Long.toString(record.offset()))); + + OutgoingKafkaRecordMetadata metadata = OutgoingKafkaRecordMetadata.builder() + .withKey(record.key()) + .withHeaders(headers) + .build(); + + return deadLetterEmitter.sendMessage(Message.of(record.value(), Metadata.of(metadata))) + .subscribeAsCompletionStage(); + } + + private static byte[] bytes(String value) { + return value.getBytes(StandardCharsets.UTF_8); } private CloudEvent validateCloudEvent(ConsumerRecord record) throws JsonProcessingException { diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/resources/application.properties b/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/resources/application.properties index 87e21b118..4ed6747fa 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/resources/application.properties +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/resources/application.properties @@ -19,6 +19,7 @@ quarkus.container-image.tag=999-SNAPSHOT # mp messaging mp.messaging.incoming.data-index-events.connector=smallrye-kafka mp.messaging.incoming.data-index-events.topic=flow-lifecycle-out +mp.messaging.incoming.data-index-events.batch=true mp.messaging.incoming.data-index-events.group.id=data-index-ingestion mp.messaging.incoming.data-index-events.auto.offset.reset=earliest mp.messaging.incoming.data-index-events.retry-attempts=2 @@ -27,9 +28,14 @@ mp.messaging.incoming.data-index-events.key.deserializer=org.apache.kafka.common mp.messaging.incoming.data-index-events.health-enabled=true mp.messaging.incoming.data-index-events.health-readiness-enabled=true -mp.messaging.incoming.data-index-events.failure-strategy=dead-letter-queue -mp.messaging.incoming.data-index-events.dead-letter-queue.topic=data-index-events-dlq -mp.messaging.incoming.data-index-events.dead-letter-queue.key.serializer=org.apache.kafka.common.serialization.StringSerializer -mp.messaging.incoming.data-index-events.dead-letter-queue.value.serializer=org.apache.kafka.common.serialization.StringSerializer +# Per-record failures are dead-lettered manually by KafkaLifecycleConsumer (batch ack/nack are +# whole-batch operations). 'fail' only triggers if a dead-letter write itself fails, so the batch +# is not committed and the failed record is not silently dropped. +mp.messaging.incoming.data-index-events.failure-strategy=fail + +mp.messaging.outgoing.data-index-events-dlq.connector=smallrye-kafka +mp.messaging.outgoing.data-index-events-dlq.topic=data-index-events-dlq +mp.messaging.outgoing.data-index-events-dlq.key.serializer=org.apache.kafka.common.serialization.StringSerializer +mp.messaging.outgoing.data-index-events-dlq.value.serializer=org.apache.kafka.common.serialization.StringSerializer quarkus.native.resources.includes=task-instance-upsert.sql,task-placeholder-workflow-insert.sql,workflow-instance-upsert.sql \ No newline at end of file From 7a73fa7e0210ea7ce16bbcd6518427d5ad9ae683 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gonzalo=20Mu=C3=B1oz?= Date: Tue, 23 Jun 2026 14:33:43 +0200 Subject: [PATCH 2/4] Add also persistence batches to execute fewer commits (#2) --- .../processor/TaskExecutionProcessor.java | 39 ++++++--- .../processor/WorkflowEventProcessor.java | 17 ++-- .../persistence/TaskPersistence.java | 31 +++++++ .../persistence/WorkflowPersistence.java | 28 +++++- .../kafka/service/KafkaLifecycleConsumer.java | 87 ++++++++++++++++--- 5 files changed, 170 insertions(+), 32 deletions(-) diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java index 54d73e472..ec51cc192 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java @@ -24,6 +24,7 @@ import org.slf4j.LoggerFactory; import java.sql.SQLException; +import java.util.List; import java.util.Objects; @Unremovable @@ -32,7 +33,7 @@ public class TaskExecutionProcessor implements EventProcessor { private static final Logger log = LoggerFactory.getLogger(TaskExecutionProcessor.class); - final TaskPersistence taskPersistence; + private final TaskPersistence taskPersistence; @Inject public TaskExecutionProcessor(TaskPersistence taskPersistence) { @@ -41,22 +42,36 @@ public TaskExecutionProcessor(TaskPersistence taskPersistence) { @Override public void process(TaskExecution event) { - Objects.requireNonNull(event, "event cannot be null"); - log.debug("Processing task: {}", event); - event.setId(generateTaskExecutionId(event)); + processBatch(List.of(event)); + } + + public void processBatch(List events) { + Objects.requireNonNull(events, "events cannot be null"); + + if (events.isEmpty()) { + return; + } + + log.info("Processing task batch size: {}", events.size()); + try { - this.taskPersistence.persist(event); - log.debug("Successfully processed the task event with ID: {}", event.getInstanceId()); + for (TaskExecution event : events) { + Objects.requireNonNull(event, "event cannot be null"); + event.setId(generateTaskExecutionId(event)); + } + + this.taskPersistence.persistBatch(events); + + log.debug("Successfully processed task batch size: {}", events.size()); } catch (SQLException e) { - log.error("Error while processing the task event: {}", event, e); - throw new ProcessEventFailedException("Failed to process the task event with instance ID: " + event.getInstanceId(), e); + log.error("Error while processing task batch size: {}", events.size(), e); + throw new ProcessEventFailedException("Failed to process task event batch", e); } } private String generateTaskExecutionId(TaskExecution taskExecutionEvent) { - // Generate deterministic ID based on instance's ID + task position - return taskExecutionEvent.getInstanceId() + - ":" + taskExecutionEvent.getTaskPosition(); + return taskExecutionEvent.getInstanceId() + + ":" + + taskExecutionEvent.getTaskPosition(); } } - diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/WorkflowEventProcessor.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/WorkflowEventProcessor.java index 7063c709b..9ac64c8b8 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/WorkflowEventProcessor.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/WorkflowEventProcessor.java @@ -24,6 +24,7 @@ import org.slf4j.LoggerFactory; import java.sql.SQLException; +import java.util.List; import java.util.Objects; @Unremovable @@ -40,13 +41,19 @@ public WorkflowEventProcessor(WorkflowPersistence workflowPersistence) { } public void process(final WorkflowInstance event) { + processBatch(List.of(event)); + } + + public void processBatch(final List events) { + log.info("Processing workflow batch size: {}", events.size()); + try { - this.workflowPersistence.persist(Objects.requireNonNull(event, "event cannot be null")); - log.debug("Successfully processed the workflow event with ID: {}", event.getId()); + workflowPersistence.persistBatch(events); + log.info("Successfully processed {} workflow events", events.size()); } catch (SQLException e) { - log.error("Error while processing the workflow event: {}", event, e); - throw new ProcessEventFailedException("Failed to process the workflow event with instance ID: " + event.getId(), e); + log.error("Error while processing workflow event batch", e); + throw new ProcessEventFailedException("Failed to process workflow event batch", e); } - } + } } diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java index a5bf7d6dc..b12df1d50 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java @@ -17,6 +17,7 @@ import java.sql.SQLException; import java.sql.Savepoint; import java.time.ZonedDateTime; +import java.util.List; import java.util.Optional; @Unremovable @@ -66,6 +67,36 @@ public void persist(TaskExecution event) throws SQLException { } } + public void persistBatch(List events) throws SQLException { + if (events == null || events.isEmpty()) { + return; + } + + log.debug("Persisting task DB batch size: {}", events.size()); + + try (Connection conn = dataSource.getConnection(); + PreparedStatement stmt = conn.prepareStatement(insertTaskUpsert)) { + + conn.setAutoCommit(false); + + try { + for (TaskExecution event : events) { + setTaskParameters(stmt, event); + stmt.addBatch(); + } + + int[] result = stmt.executeBatch(); + conn.commit(); + + log.debug("Committed task DB batch size: {}, executeBatch result length: {}", + events.size(), result.length); + } catch (SQLException e) { + conn.rollback(); + throw e; + } + } + } + private void tryInsertTask(TaskExecution event, Connection conn) throws SQLException { try (PreparedStatement stmt = conn.prepareStatement(insertTaskUpsert)) { setTaskParameters(stmt, event); diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/WorkflowPersistence.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/WorkflowPersistence.java index 6a4646e46..ccbae10a7 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/WorkflowPersistence.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/WorkflowPersistence.java @@ -17,6 +17,7 @@ import java.sql.SQLException; import java.sql.Types; import java.time.ZonedDateTime; +import java.util.List; import java.util.Optional; @Unremovable @@ -58,6 +59,31 @@ public void persist(WorkflowInstance event) throws SQLException { } } + public void persistBatch(List events) throws SQLException { + if (events == null || events.isEmpty()) { + return; + } + + try (Connection conn = dataSource.getConnection(); + PreparedStatement stmt = conn.prepareStatement(insertWorkflowUpsert)) { + + conn.setAutoCommit(false); + + try { + for (WorkflowInstance event : events) { + setWorkflowParameters(stmt, event); + stmt.addBatch(); + } + + stmt.executeBatch(); + conn.commit(); + } catch (SQLException e) { + conn.rollback(); + throw e; + } + } + } + private void setWorkflowParameters(PreparedStatement stmt, WorkflowInstance event) throws SQLException { stmt.setString(1, event.getId()); stmt.setString(2, event.getNamespace()); @@ -105,4 +131,4 @@ private String toJsonString(JsonNode node) { } } -} \ No newline at end of file +} diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java index e15eaa43c..176562ca3 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java @@ -28,12 +28,16 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.header.internals.RecordHeaders; +import org.eclipse.microprofile.config.inject.ConfigProperty; import org.eclipse.microprofile.reactive.messaging.Channel; import org.eclipse.microprofile.reactive.messaging.Incoming; import org.eclipse.microprofile.reactive.messaging.Message; import org.eclipse.microprofile.reactive.messaging.Metadata; import org.kubesmarts.logic.dataindex.ingestion.kafka.processor.EventProcessor; import org.kubesmarts.logic.dataindex.ingestion.kafka.processor.ProcessEventFailedException; +import org.kubesmarts.logic.dataindex.ingestion.kafka.processor.WorkflowEventProcessor; +import org.kubesmarts.logic.dataindex.ingestion.kafka.processor.TaskExecutionProcessor; + import org.kubesmarts.logic.dataindex.model.LifecycleEventUtils; import org.kubesmarts.logic.dataindex.model.TaskExecution; import org.kubesmarts.logic.dataindex.model.WorkflowInstance; @@ -46,45 +50,57 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; + @ApplicationScoped public class KafkaLifecycleConsumer { private static final Logger log = LoggerFactory.getLogger(KafkaLifecycleConsumer.class); - @Inject - ObjectMapper jackson; + private final WorkflowEventProcessor workflowEventProcessor; + private final TaskExecutionProcessor taskExecutionProcessor; @Inject - EventProcessor workflowEventProcessor; - + public KafkaLifecycleConsumer(WorkflowEventProcessor workflowEventProcessor, + TaskExecutionProcessor taskExecutionProcessor) + { + this.workflowEventProcessor = workflowEventProcessor; + this.taskExecutionProcessor = taskExecutionProcessor; + } + @Inject - EventProcessor taskExecutionProcessor; + ObjectMapper jackson; @Inject @Channel("data-index-events-dlq") MutinyEmitter deadLetterEmitter; + @ConfigProperty(name = "data-index.ingestion.db-batch-size", defaultValue = "1000") + int dbBatchSize; + @Incoming("data-index-events") public CompletionStage consumeLifecycleEvent(Message> records) { List> deadLetterSends = new ArrayList<>(); - + + List workflowInstances = new ArrayList<>(); + List taskExecutions = new ArrayList<>(); + for (ConsumerRecord record : records.getPayload()) { try { CloudEvent cloudEvent = validateCloudEvent(record); - + JsonCloudEventData cloudEventData = (JsonCloudEventData) cloudEvent.getData(); if (cloudEventData == null || cloudEventData.getNode() == null) { throw new IllegalArgumentException("The CloudEvent data node consumed at offset %s from partition %s is null or empty." .formatted(record.offset(), record.partition())); } - + Class eventClass = LifecycleEventUtils.getEventClass(cloudEvent.getType()); Object data = jackson.convertValue(cloudEventData.getNode(), eventClass); - + if (data instanceof TaskCEData taskData) { - handleTaskEvent(cloudEvent, taskData); + taskExecutions.add(mapTaskEvent(cloudEvent, taskData)); } else if (data instanceof WorkflowCEData workflowData) { - handleWorkflowEvent(cloudEvent, workflowData); + workflowInstances.add(mapWorkflowEvent(cloudEvent, workflowData)); } else { throw new IllegalArgumentException("Unsupported event type '%s' consumed at offset %s from partition %s." .formatted(cloudEvent.getType(), record.offset(), record.partition())); @@ -95,14 +111,57 @@ public CompletionStage consumeLifecycleEvent(Message chunk : partition(workflowInstances, dbBatchSize)) { + workflowEventProcessor.processBatch(chunk); + } + + for (List chunk : partition(taskExecutions, dbBatchSize)) { + taskExecutionProcessor.processBatch(chunk); + } + } catch (Exception e) { + log.error("Failed to persist Kafka batch. Nacking the batch so it can be retried.", e); + return records.nack(e); + } + CompletableFuture[] pending = deadLetterSends.stream() .map(CompletionStage::toCompletableFuture) .toArray(CompletableFuture[]::new); - + return CompletableFuture.allOf(pending).thenCompose(ignored -> records.ack()); } + + private static List> partition(List list, int size) { + List> chunks = new ArrayList<>(); + + for (int i = 0; i < list.size(); i += size) { + chunks.add(list.subList(i, Math.min(i + size, list.size()))); + } + + return chunks; + } + + private TaskExecution mapTaskEvent(CloudEvent cloudEvent, TaskCEData data) { + try { + return Mapper.mapTaskExecutionEvent(cloudEvent, data, jackson); + } catch (Exception e) { + log.error("Error while mapping CloudEvent (task) with ID: {}", cloudEvent.getId(), e); + throw new ProcessEventFailedException("Failed to map CloudEvent with ID: " + cloudEvent.getId(), e); + } + } + + private WorkflowInstance mapWorkflowEvent(CloudEvent cloudEvent, WorkflowCEData data) { + try { + return Mapper.mapWorkflowInstanceEvent(cloudEvent, data, jackson); + } catch (Exception e) { + log.error("Error while mapping CloudEvent (workflow) with ID: {}", data.getName(), e); + throw new ProcessEventFailedException("Failed to map CloudEvent with ID: " + data.getName(), e); + } + } private CompletionStage sendToDeadLetterQueue(ConsumerRecord record, Exception cause) { RecordHeaders headers = new RecordHeaders(); From 52297dd1921999c0a4cb706af2c091c6af2f9679 Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Wed, 24 Jun 2026 13:10:32 -0300 Subject: [PATCH 3/4] Recover placeholder workflows in batched task persistence The batched task upsert (persistBatch) cannot create the placeholder workflow needed when a task event arrives before its workflow, so the whole batch failed with a foreign key violation and was retried forever. On a FK violation, roll back and fall back to per-record persistence, which already creates placeholders idempotently. Co-Authored-By: Claude Opus 4.8 --- .../persistence/TaskPersistence.java | 29 +++++++++++++++++-- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java index b12df1d50..d3d81a769 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java @@ -84,17 +84,40 @@ public void persistBatch(List events) throws SQLException { setTaskParameters(stmt, event); stmt.addBatch(); } - + int[] result = stmt.executeBatch(); conn.commit(); - + log.debug("Committed task DB batch size: {}, executeBatch result length: {}", events.size(), result.length); } catch (SQLException e) { conn.rollback(); - throw e; + // A task may arrive before its workflow. The batch upsert cannot create the + // missing placeholder workflow, so fall back to per-record persistence which + // recovers from foreign key violations individually. + if (isForeignKeyViolation(e)) { + log.debug("Task batch hit foreign key violation; falling back to per-record persistence"); + persistEachIndividually(events); + } else { + throw e; + } + } + } + } + + private void persistEachIndividually(List events) throws SQLException { + for (TaskExecution event : events) { + persist(event); + } + } + + private boolean isForeignKeyViolation(SQLException e) { + for (SQLException current = e; current != null; current = current.getNextException()) { + if (INVALID_FOREIGN_KEY.equals(current.getSQLState())) { + return true; } } + return false; } private void tryInsertTask(TaskExecution event, Connection conn) throws SQLException { From eebb7192be0073958a6ee3dd0c4593be807da62b Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Wed, 24 Jun 2026 22:07:26 -0300 Subject: [PATCH 4/4] Do a clean up Signed-off-by: Matheus Cruz --- .../ingestion/kafka/processor/TaskExecutionProcessor.java | 2 +- .../ingestion/kafka/processor/WorkflowEventProcessor.java | 4 ++-- .../ingestion/kafka/service/KafkaLifecycleConsumer.java | 3 --- 3 files changed, 3 insertions(+), 6 deletions(-) diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java index ec51cc192..9d3290980 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java @@ -52,7 +52,7 @@ public void processBatch(List events) { return; } - log.info("Processing task batch size: {}", events.size()); + log.debug("Processing task batch size: {}", events.size()); try { for (TaskExecution event : events) { diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/WorkflowEventProcessor.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/WorkflowEventProcessor.java index 9ac64c8b8..40422aee9 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/WorkflowEventProcessor.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/WorkflowEventProcessor.java @@ -45,11 +45,11 @@ public void process(final WorkflowInstance event) { } public void processBatch(final List events) { - log.info("Processing workflow batch size: {}", events.size()); + log.debug("Processing workflow batch size: {}", events.size()); try { workflowPersistence.persistBatch(events); - log.info("Successfully processed {} workflow events", events.size()); + log.debug("Successfully processed {} workflow events", events.size()); } catch (SQLException e) { log.error("Error while processing workflow event batch", e); throw new ProcessEventFailedException("Failed to process workflow event batch", e); diff --git a/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java b/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java index 176562ca3..0ce368cf4 100644 --- a/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java +++ b/data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/KafkaLifecycleConsumer.java @@ -101,9 +101,6 @@ public CompletionStage consumeLifecycleEvent(Message