ayush03dev · May 23, 2026 · May 23, 2026 · May 23, 2026
diff --git a/...-service/src/main/java/com/ayushlabs/codebox/agentservice/service/JobExecutionRunner.java b/...-service/src/main/java/com/ayushlabs/codebox/agentservice/service/JobExecutionRunner.java
@@ -2,8 +2,6 @@
 
 import com.ayushlabs.codebox.agentservice.dto.AgentJobDispatchRequest;
 import com.ayushlabs.codebox.agentservice.entity.ExecutionStatus;
-import com.ayushlabs.codebox.agentservice.entity.JobExecution;
-import com.ayushlabs.codebox.agentservice.repository.JobExecutionRepository;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
@@ -14,29 +12,26 @@
 @Component
 public class JobExecutionRunner {
 
-    private final JobExecutionRepository jobExecutionRepository;
+    private final JobExecutionUpdater jobExecutionUpdater;
     private final ContainerExecutor containerExecutor;
     private final LogServiceClient logServiceClient;
     private final Clock clock;
 
     public JobExecutionRunner(
-            JobExecutionRepository jobExecutionRepository,
+            JobExecutionUpdater jobExecutionUpdater,
             ContainerExecutor containerExecutor,
             LogServiceClient logServiceClient
     ) {
-        this.jobExecutionRepository = jobExecutionRepository;
+        this.jobExecutionUpdater = jobExecutionUpdater;
         this.containerExecutor = containerExecutor;
         this.logServiceClient = logServiceClient;
         this.clock = Clock.systemUTC();
     }
 
     @Async
     public void executeAsync(UUID executionId, AgentJobDispatchRequest request) {
-        JobExecution execution = jobExecutionRepository.findById(executionId).orElseThrow();
-
         String image = containerExecutor.resolveImage(request.language(), request.runtime());
-        execution.start(image, clock.instant());
-        jobExecutionRepository.save(execution);
+        jobExecutionUpdater.markRunning(executionId, image, clock.instant());
 
         logServiceClient.startSession(executionId, request.jobId(), request.userId());
 
@@ -54,7 +49,12 @@ public void executeAsync(UUID executionId, AgentJobDispatchRequest request) {
                 ? ExecutionStatus.TIMED_OUT
                 : result.exitCode() == 0 ? ExecutionStatus.SUCCEEDED : ExecutionStatus.FAILED;
 
-        execution.complete(status, result.exitCode(), result.logs(), Instant.now(clock));
-        jobExecutionRepository.save(execution);
+        jobExecutionUpdater.markComplete(
+                executionId,
+                status,
+                result.exitCode(),
+                result.logs(),
+                Instant.now(clock)
+        );
     }
 }
diff --git a/...service/src/main/java/com/ayushlabs/codebox/agentservice/service/JobExecutionService.java b/...service/src/main/java/com/ayushlabs/codebox/agentservice/service/JobExecutionService.java
@@ -7,6 +7,10 @@
 import com.ayushlabs.codebox.agentservice.repository.JobExecutionRepository;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import java.util.UUID;
 
 @Service
 public class JobExecutionService {
@@ -31,7 +35,21 @@ public AgentJobDispatchResponse accept(AgentJobDispatchRequest request) {
                 .build();
 
         JobExecution saved = jobExecutionRepository.save(execution);
-        jobExecutionRunner.executeAsync(saved.getId(), request);
+        executeAfterCommit(saved.getId(), request);
         return new AgentJobDispatchResponse(saved.getId(), saved.getStatus().name(), "Job accepted for execution");
     }
+
+    private void executeAfterCommit(UUID executionId, AgentJobDispatchRequest request) {
+        if (!TransactionSynchronizationManager.isSynchronizationActive()) {
+            jobExecutionRunner.executeAsync(executionId, request);
+            return;
+        }
+
+        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+            @Override
+            public void afterCommit() {
+                jobExecutionRunner.executeAsync(executionId, request);
+            }
+        });
+    }
 }
diff --git a/...service/src/main/java/com/ayushlabs/codebox/agentservice/service/JobExecutionUpdater.java b/...service/src/main/java/com/ayushlabs/codebox/agentservice/service/JobExecutionUpdater.java
@@ -0,0 +1,38 @@
+package com.ayushlabs.codebox.agentservice.service;
+
+import com.ayushlabs.codebox.agentservice.entity.ExecutionStatus;
+import com.ayushlabs.codebox.agentservice.entity.JobExecution;
+import com.ayushlabs.codebox.agentservice.repository.JobExecutionRepository;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.UUID;
+
+@Service
+public class JobExecutionUpdater {
+
+    private final JobExecutionRepository jobExecutionRepository;
+
+    public JobExecutionUpdater(JobExecutionRepository jobExecutionRepository) {
+        this.jobExecutionRepository = jobExecutionRepository;
+    }
+
+    @Transactional
+    public void markRunning(UUID executionId, String containerImage, Instant startedAt) {
+        JobExecution execution = jobExecutionRepository.findById(executionId).orElseThrow();
+        execution.start(containerImage, startedAt);
+    }
+
+    @Transactional
+    public void markComplete(
+            UUID executionId,
+            ExecutionStatus status,
+            int exitCode,
+            String logs,
+            Instant completedAt
+    ) {
+        JobExecution execution = jobExecutionRepository.findById(executionId).orElseThrow();
+        execution.complete(status, exitCode, logs, completedAt);
+    }
+}
diff --git a/services/job-service/pom.xml b/services/job-service/pom.xml
@@ -30,6 +30,11 @@
             <artifactId>spring-boot-starter-data-jpa</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>
@@ -68,6 +73,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
     <build>

diff --git a/services/job-service/src/main/java/com/ayushlabs/codebox/jobservice/config/KafkaConfig.java b/services/job-service/src/main/java/com/ayushlabs/codebox/jobservice/config/KafkaConfig.java
@@ -0,0 +1,52 @@
+package com.ayushlabs.codebox.jobservice.config;
+
+import com.ayushlabs.codebox.jobservice.kafka.JobCreatedEvent;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@EnableKafka
+@Configuration
+public class KafkaConfig {
+
+    @Bean
+    public ConsumerFactory<String, JobCreatedEvent> jobCreatedConsumerFactory(
+            @Value("${spring.kafka.bootstrap-servers}") String bootstrapServers,
+            @Value("${codebox.kafka.consumer.group-id}") String groupId
+    ) {
+        Map<String, Object> properties = new HashMap<>();
+        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        properties.put(JsonDeserializer.TRUSTED_PACKAGES, "com.ayushlabs.codebox.jobservice.kafka");
+        properties.put(JsonDeserializer.VALUE_DEFAULT_TYPE, JobCreatedEvent.class.getName());
+
+        return new DefaultKafkaConsumerFactory<>(
+                properties,
+                new StringDeserializer(),
+                new JsonDeserializer<>(JobCreatedEvent.class, false)
+        );
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, JobCreatedEvent> jobCreatedKafkaListenerContainerFactory(
+            ConsumerFactory<String, JobCreatedEvent> jobCreatedConsumerFactory
+    ) {
+        ConcurrentKafkaListenerContainerFactory<String, JobCreatedEvent> factory =
+                new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(jobCreatedConsumerFactory);
+        return factory;
+    }
+}
diff --git a/.../job-service/src/main/java/com/ayushlabs/codebox/jobservice/kafka/JobCreatedConsumer.java b/.../job-service/src/main/java/com/ayushlabs/codebox/jobservice/kafka/JobCreatedConsumer.java
@@ -0,0 +1,29 @@
+package com.ayushlabs.codebox.jobservice.kafka;
+
+import com.ayushlabs.codebox.jobservice.service.JobDispatchService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Component
+public class JobCreatedConsumer {
+
+    private static final Logger log = LoggerFactory.getLogger(JobCreatedConsumer.class);
+
+    private final JobDispatchService jobDispatchService;
+
+    public JobCreatedConsumer(JobDispatchService jobDispatchService) {
+        this.jobDispatchService = jobDispatchService;
+    }
+
+    @KafkaListener(
+            topics = "${codebox.kafka.topics.job-created}",
+            groupId = "${codebox.kafka.consumer.group-id}",
+            containerFactory = "jobCreatedKafkaListenerContainerFactory"
+    )
+    public void onJobCreated(JobCreatedEvent event) {
+        log.info("event=job_created_consumed jobId={} eventId={}", event.jobId(), event.eventId());
+        jobDispatchService.dispatchJobAsync(event.jobId());
+    }
+}
diff --git a/...ces/job-service/src/main/java/com/ayushlabs/codebox/jobservice/kafka/JobCreatedEvent.java b/...ces/job-service/src/main/java/com/ayushlabs/codebox/jobservice/kafka/JobCreatedEvent.java
@@ -0,0 +1,19 @@
+package com.ayushlabs.codebox.jobservice.kafka;
+
+import com.ayushlabs.codebox.jobservice.entity.JobStatus;
+
+import java.time.Instant;
+import java.util.UUID;
+
+public record JobCreatedEvent(
+        UUID eventId,
+        UUID jobId,
+        UUID userId,
+        String language,
+        String runtime,
+        JobStatus status,
+        int retryCount,
+        Instant createdAt,
+        Instant occurredAt
+) {
+}
diff --git a/...s/job-service/src/main/java/com/ayushlabs/codebox/jobservice/kafka/JobEventPublisher.java b/...s/job-service/src/main/java/com/ayushlabs/codebox/jobservice/kafka/JobEventPublisher.java
@@ -0,0 +1,40 @@
+package com.ayushlabs.codebox.jobservice.kafka;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+public class JobEventPublisher {
+
+    private static final Logger log = LoggerFactory.getLogger(JobEventPublisher.class);
+
+    private final KafkaTemplate<String, JobCreatedEvent> kafkaTemplate;
+    private final String jobCreatedTopic;
+
+    public JobEventPublisher(
+            KafkaTemplate<String, JobCreatedEvent> kafkaTemplate,
+            @Value("${codebox.kafka.topics.job-created}") String jobCreatedTopic
+    ) {
+        this.kafkaTemplate = kafkaTemplate;
+        this.jobCreatedTopic = jobCreatedTopic;
+    }
+
+    public void publishJobCreated(JobCreatedEvent event) {
+        kafkaTemplate.send(jobCreatedTopic, event.jobId().toString(), event)
+                .whenComplete((result, exception) -> {
+                    if (exception != null) {
+                        log.error(
+                                "event=job_created_publish_failed jobId={} topic={}",
+                                event.jobId(),
+                                jobCreatedTopic,
+                                exception
+                        );
+                        return;
+                    }
+                    log.info("event=job_created_published jobId={} topic={}", event.jobId(), jobCreatedTopic);
+                });
+    }
+}