system_report.md

Distributed MapReduce System Report

---

Table of Contents

1. What is this system?
2. Project Structure
3. Parent POM & Dependency Management
4. common — Shared Interfaces & DTOs
5. ui-service — The Public API Gateway
6. manager-service — The Brain
7. worker — The Compute Pod
8. cli — The Command-Line Tool
9. examples/wordcount — Sample Job
10. External Infrastructure
11. Authentication Flows
12. File Upload Flow
13. Full Job Lifecycle
14. Failure Handling & Retries
15. Dockerfiles
16. Kubernetes Setup
17. Docker Compose (local dev)
18. Database Schema
19. Monitoring & Health
20. Complete End-to-End Flow

---

1. What is this system?

Ok so at a high level this is a cloud-native distributed MapReduce platform. The classic example people use for MapReduce is word count — you have a huge text file and want to count how many times each word appears. Instead of doing that on one machine (slow for big files), you split the work across many machines. That's MapReduce: Map phase splits the work, Reduce phase combines the results.

The interesting part of this project is how it's all implemented:

• It's not Hadoop. It doesn't use HDFS. It uses MinIO (self-hosted S3) for storage.
• Workers aren't long-lived servers. Each map/reduce task is its own Kubernetes batch Job — a pod that starts, does its thing, and exits.
• Users bring their own code. They upload a JAR file that implements the Mapper and Reducer interfaces. The system loads it at runtime via Java's ServiceLoader mechanism.
• Auth is done with Keycloak — real OAuth2/OIDC, real JWTs, real JWKS validation.
• The user interacts via a CLI tool (mr) that has an interactive shell mode.

The whole thing runs in Kubernetes. The manager service dynamically creates worker pods via the Fabric8 Kubernetes client. When a job finishes, you get back presigned MinIO URLs to download the results.

---

2. Project Structure

This is a Maven multi-module project. The root pom.xml declares all the modules and manages shared dependency versions via a BOM-style <dependencyManagement> block.

distributed/
│
├── pom.xml                          ← parent POM, manages all versions
│
├── common/                          ← shared library (NOT a runnable service)
│   └── src/main/java/gr/tuc/distributed/common/
│       ├── api/
│       │   ├── Mapper.java          ← interface users implement
│       │   └── Reducer.java         ← interface users implement
│       ├── dto/
│       │   ├── JobSubmitRequest.java
│       │   ├── JobStatusResponse.java
│       │   ├── TaskStatusUpdate.java
│       │   └── FileUploadResponse.java
│       └── enums/
│           ├── JobStatus.java
│           ├── TaskStatus.java
│           └── TaskType.java
│
├── ui-service/                      ← public REST API, auth, forwards to manager
│   ├── Dockerfile
│   └── src/main/java/gr/tuc/distributed/ui/
│       ├── UiApplication.java
│       ├── config/
│       │   ├── SecurityConfig.java
│       │   ├── WebClientConfig.java
│       │   └── OpenApiConfig.java
│       ├── controller/
│       │   ├── AuthController.java
│       │   ├── JobController.java
│       │   ├── DataController.java
│       │   ├── CodeController.java
│       │   └── AdminController.java
│       └── client/
│           ├── ManagerClient.java
│           └── KeycloakAdminClient.java
│
├── manager-service/                 ← orchestration, K8s, DB, MinIO
│   ├── Dockerfile
│   └── src/main/java/gr/tuc/distributed/manager/
│       ├── ManagerApplication.java
│       ├── config/
│       │   ├── SecurityConfig.java
│       │   ├── MinioConfig.java
│       │   └── KubernetesConfig.java
│       ├── controller/
│       │   ├── InternalJobController.java
│       │   ├── InternalTaskController.java
│       │   └── InternalFileController.java
│       ├── service/
│       │   ├── JobOrchestrationService.java
│       │   └── FileService.java
│       ├── k8s/
│       │   └── KubernetesJobLauncher.java
│       ├── minio/
│       │   ├── MinioStorageService.java
│       │   └── MinioOperationException.java
│       ├── entity/
│       │   ├── Job.java
│       │   ├── Task.java
│       │   └── FileMetadata.java
│       ├── repository/
│       │   ├── JobRepository.java
│       │   ├── TaskRepository.java
│       │   └── FileMetadataRepository.java
│       └── scheduler/
│           └── HeartbeatWatchdog.java
│
├── worker/                          ← stateless batch pod
│   ├── Dockerfile
│   └── src/main/java/gr/tuc/distributed/worker/
│       ├── WorkerApplication.java
│       ├── runner/
│       │   └── WorkerRunner.java
│       ├── minio/
│       │   └── WorkerMinioClient.java
│       └── reporter/
│           └── ManagerReporter.java
│
├── cli/                             ← fat JAR CLI tool
│   ├── mr                           ← shell launcher script
│   └── src/main/java/gr/tuc/distributed/cli/
│       ├── Main.java
│       ├── MapReduceCli.java
│       ├── api/
│       │   ├── ApiClient.java
│       │   └── ApiException.java
│       ├── command/
│       │   ├── LoginCommand.java
│       │   ├── LogoutCommand.java
│       │   ├── RegisterCommand.java
│       │   ├── WhoamiCommand.java
│       │   ├── UploadCommand.java
│       │   ├── SubmitCommand.java
│       │   ├── RunCommand.java
│       │   ├── JobsCommand.java
│       │   ├── StatusCommand.java
│       │   ├── CancelCommand.java
│       │   └── ResultsCommand.java
│       ├── config/
│       │   ├── Session.java
│       │   └── SessionStore.java
│       └── util/
│           ├── Out.java
│           └── MultipartBodyPublisher.java
│
├── examples/
│   └── wordcount/                   ← sample MapReduce job
│       └── src/main/java/gr/tuc/distributed/examples/wordcount/
│           ├── WordCountMapper.java
│           └── WordCountReducer.java
│
├── k8s/                             ← all Kubernetes manifests
│   ├── namespace/namespace.yaml
│   ├── postgres/
│   ├── minio/
│   ├── keycloak/
│   ├── manager/
│   └── ui/
│
└── docker-compose.yml               ← local dev setup

---

3. Parent POM & Dependency Management

File: pom.xml (root)

• groupId: gr.tuc.distributed
• artifactId: mapreduce-parent
• version: 1.0.0-SNAPSHOT
• packaging: pom

The parent POM's main job is version management so you don't repeat versions in every child module. Key things it controls:

<properties>
    <java.version>21</java.version>
    <spring-boot.version>3.3.4</spring-boot.version>
    <fabric8.version>6.13.1</fabric8.version>
    <minio.version>8.5.11</minio.version>
    <mapstruct.version>1.5.5.Final</mapstruct.version>
    <lombok.version>1.18.34</lombok.version>
    <flyway.version>10.15.0</flyway.version>
    <testcontainers.version>1.20.1</testcontainers.version>
</properties>

The <dependencyManagement> section imports the Spring Boot BOM and the Spring Cloud BOM, plus pins Fabric8, MinIO, Flyway, MapStruct, and TestContainers. This means child POMs just say <artifactId>kubernetes-client</artifactId> without a version number — it comes from here.

Build plugins managed at the parent level:

• spring-boot-maven-plugin — for packaging runnable JARs
• maven-compiler-plugin — with annotation processors for Lombok + MapStruct (they run at compile time)

---

4. common — Shared Interfaces & DTOs

This module is a library — it produces a JAR that the other modules depend on. It never runs standalone.

Why does it exist? Because manager-service, worker, and cli all need to agree on what a JobStatus looks like, what TaskStatusUpdate contains, what Mapper.map() signature is, etc. Keeping these in a shared module means you only define them once.

4.1 The Mapper and Reducer interfaces

// gr.tuc.distributed.common.api.Mapper
public interface Mapper {
    void map(String inputKey, String line, BiConsumer<String, String> emit);
}

// gr.tuc.distributed.common.api.Reducer
public interface Reducer {
    void reduce(String key, List<String> values, BiConsumer<String, String> emit);
}

The emit parameter is a BiConsumer<String, String> — it's basically a callback. When your mapper finds a word, instead of returning a list, you just call emit.accept("word", "1"). This is cleaner than returning a List<Map.Entry<>> because the framework can decide how to buffer/flush without the user needing to manage collections.

These interfaces are loaded at runtime via ServiceLoader (explained in detail in the worker section).

4.2 Enums

// JobStatus.java
public enum JobStatus {
    INITIALIZING,   // job created, not started yet
    MAP_PHASE,      // map workers running
    REDUCE_PHASE,   // reduce workers running
    COMPLETED,      // all done, results available
    FAILED,         // something went wrong permanently
    CANCELLED       // user cancelled it
}

// TaskStatus.java
public enum TaskStatus {
    IDLE,           // task created, worker not launched yet
    IN_PROGRESS,    // worker pod is running
    COMPLETED,      // worker finished successfully
    FAILED          // worker failed, may be retried
}

// TaskType.java
public enum TaskType {
    MAP,
    REDUCE
}

4.3 DTOs (Data Transfer Objects)

These are the objects that get serialized to/from JSON when services talk to each other or when the CLI talks to the API.

JobSubmitRequest — what you send when submitting a job:

@Data
public class JobSubmitRequest {
    @NotBlank
    private String dataId;      // UUID of uploaded data file

    @NotBlank
    private String codeId;      // UUID of uploaded code JAR

    @Min(1)
    private int numReducers;    // how many reduce workers to use
}

JobStatusResponse — what you get back when polling a job:

@Data @Builder @NoArgsConstructor @AllArgsConstructor
public class JobStatusResponse {
    private UUID jobId;
    private JobStatus status;
    private List<String> outputUrls;   // presigned download URLs (only when COMPLETED)
    private String errorMessage;       // only set if FAILED
    private Instant createdAt;
    private Instant updatedAt;
}

TaskStatusUpdate — what a worker sends to the manager when its status changes:

@Data
public class TaskStatusUpdate {
    @NotNull
    private TaskStatus status;       // IN_PROGRESS, COMPLETED, or FAILED

    private String outputLocation;   // MinIO path to output (set when COMPLETED)
    private String errorMessage;     // error description (set when FAILED)
}

FileUploadResponse — what you get back after uploading a file:

@Data @Builder @NoArgsConstructor @AllArgsConstructor
public class FileUploadResponse {
    private String id;           // UUID of the file_metadata record
    private String storagePath;  // MinIO object key
}

Dependencies for common: Lombok (for @Data, @Builder, etc.), Jackson annotations (for JSON), Jakarta Validation API (for @NotBlank, @Min).

---

5. ui-service — The Public API Gateway

Port: 8080 (exposed as NodePort 30080 in Kubernetes)

Role: This is the only service the outside world ever talks to. It handles authentication (login, register), then proxies everything else to the manager service. It also validates JWT tokens on every protected endpoint.

5.1 Dependencies

spring-boot-starter-web             → REST server
spring-boot-starter-security        → security filter chain
spring-boot-starter-oauth2-resource-server → JWT validation against Keycloak JWKS
spring-boot-starter-validation      → @Valid on request bodies
spring-boot-starter-actuator        → /actuator/health for K8s probes
micrometer-registry-prometheus      → metrics at /actuator/prometheus
springdoc-openapi-starter-webmvc-ui → Swagger UI at /swagger-ui.html
lombok                              → @Data, @RequiredArgsConstructor, etc.
common (internal)                   → shared DTOs

5.2 Configuration (application.yml)

server:
  port: 8080

spring:
  security:
    oauth2:
      resourceserver:
        jwt:
          issuer-uri: ${KEYCLOAK_ISSUER_URI}    # e.g. http://keycloak:8080/realms/mapreduce
          jwk-set-uri: ${KEYCLOAK_JWK_SET_URI}  # same base + /protocol/openid-connect/certs
  servlet:
    multipart:
      max-file-size: 1GB
      max-request-size: 1GB

manager:
  base-url: ${MANAGER_BASE_URL:http://manager-service:8081}

keycloak:
  realm: mapreduce
  admin:
    base-url: ${KEYCLOAK_ADMIN_BASE_URL}
    username: ${KEYCLOAK_ADMIN_USERNAME}
    password: ${KEYCLOAK_ADMIN_PASSWORD}

5.3 Security Configuration (SecurityConfig.java)

@Configuration
@EnableWebSecurity
public class SecurityConfig {

    @Bean
    public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
        return http
            .csrf(csrf -> csrf.disable())
            .sessionManagement(s -> s.sessionCreationPolicy(STATELESS))
            .authorizeHttpRequests(auth -> auth
                // public endpoints
                .requestMatchers("/actuator/health/**").permitAll()
                .requestMatchers("/swagger-ui/**", "/v3/api-docs/**").permitAll()
                .requestMatchers("/api/auth/**").permitAll()
                // admin-only
                .requestMatchers("/api/v1/admin/**").hasRole("admin")
                // everything else needs a valid JWT
                .anyRequest().authenticated()
            )
            .oauth2ResourceServer(oauth2 -> oauth2
                .jwt(jwt -> jwt
                    .decoder(jwtDecoder())
                    .jwtAuthenticationConverter(keycloakJwtConverter())
                )
            )
            .build();
    }

The keycloakJwtConverter() bean is important — Keycloak puts roles inside realm_access.roles in the JWT, but Spring Security expects ROLE_ prefixed authorities. The converter reads realm_access.roles and maps each one to ROLE_<rolename>.

5.4 Controllers

AuthController — /api/auth/**

This is the only controller that doesn't require a JWT to call.

@RestController
@RequestMapping("/api/auth")
@RequiredArgsConstructor
public class AuthController {

    private final KeycloakAdminClient keycloakAdminClient;

    // Login: exchange username+password for a JWT
    @PostMapping("/token")
    public ResponseEntity<?> login(@RequestBody LoginRequest body) { ... }

    // Check if a username is already taken before registering
    @GetMapping("/check-username")
    public ResponseEntity<Map<String, Boolean>> checkUsername(@RequestParam String username) { ... }

    // Create a new account
    @PostMapping("/register")
    public ResponseEntity<?> register(@RequestBody RegisterRequest body) { ... }

    @Data static class LoginRequest  { String username; String password; }
    @Data static class RegisterRequest { String username; String password; }
}

The login method calls Keycloak's token endpoint directly:

POST http://keycloak:8080/realms/mapreduce/protocol/openid-connect/token
Content-Type: application/x-www-form-urlencoded

grant_type=password&client_id=mapreduce-cli&username=alice&password=secret123

Keycloak responds with a JSON body containing access_token, expires_in, token_type, etc. The UI service extracts those and returns them to the CLI.

JobController — /api/v1/jobs

@RestController
@RequestMapping("/api/v1/jobs")
@RequiredArgsConstructor
public class JobController {

    private final ManagerClient managerClient;

    @GetMapping
    public List<JobStatusResponse> listJobs(HttpServletRequest request) { ... }

    @PostMapping
    public ResponseEntity<Map<String, String>> submitJob(
            @Valid @RequestBody JobSubmitRequest body,
            HttpServletRequest request) { ... }

    @GetMapping("/{jobId}")
    public JobStatusResponse getJobStatus(
            @PathVariable UUID jobId,
            HttpServletRequest request) { ... }

    @DeleteMapping("/{jobId}")
    public ResponseEntity<Void> cancelJob(
            @PathVariable UUID jobId,
            HttpServletRequest request) { ... }
}

Every method extracts the Authorization header from the incoming HttpServletRequest and forwards it to ManagerClient. The manager service then validates the same JWT again on its end.

DataController — /api/v1/data

CodeController — /api/v1/code

Both are simple proxies — they receive a multipart/form-data POST with a file, forward it to the manager service with the same auth header, and return the FileUploadResponse.

AdminController — /api/v1/admin/**

Requires ROLE_admin. Exposes:

• POST /api/v1/admin/users — create a user (calls Keycloak admin API)
• DELETE /api/v1/admin/users/{userId} — delete a user

@Data
static class CreateUserRequest {
    @NotBlank String username;
    @NotBlank String email;
    @NotBlank String password;
}

5.5 Client Classes

ManagerClient

This is how the UI service talks to the manager service. It's built on Spring's RestClient (new in Spring 6, replaces RestTemplate).

@Component
public class ManagerClient {

    private final RestClient managerClient;   // injected via @Qualifier

    public FileUploadResponse uploadData(MultipartFile file, String authHeader) {
        // POST /internal/files/data (multipart/form-data)
    }

    public FileUploadResponse uploadCode(MultipartFile file, String authHeader) {
        // POST /internal/files/code (multipart/form-data)
    }

    public List<JobStatusResponse> listJobs(String authHeader) {
        // GET /internal/jobs
    }

    public String submitJob(JobSubmitRequest request, String authHeader) {
        // POST /internal/jobs → returns jobId string
    }

    public JobStatusResponse getJobStatus(UUID jobId, String authHeader) {
        // GET /internal/jobs/{jobId}
    }

    public void cancelJob(UUID jobId, String authHeader) {
        // DELETE /internal/jobs/{jobId}
    }
}

The RestClient is configured in WebClientConfig with the manager base URL from application.yml, so the client methods just specify the path.

KeycloakAdminClient

This is how the UI service creates/manages Keycloak users. It uses the Keycloak Admin REST API (not a dedicated SDK).

@Component
@Slf4j
public class KeycloakAdminClient {

    @Value("${keycloak.admin.base-url}") private String baseUrl;
    @Value("${keycloak.admin.username}") private String adminUsername;
    @Value("${keycloak.admin.password}") private String adminPassword;
    @Value("${keycloak.realm}")          private String realm;

    // Gets a short-lived admin token from the master realm
    private String adminToken() {
        // POST http://keycloak:8080/realms/master/protocol/openid-connect/token
        // grant_type=password, client_id=admin-cli, username=admin, password=...
    }

    // Returns true if the username is NOT taken
    public boolean isUsernameAvailable(String username) {
        // GET /admin/realms/{realm}/users?username={username}&exact=true
        // returns false if response list is non-empty
    }

    // Creates the user and sets their password
    public void createUser(String username, String password) {
        // POST /admin/realms/{realm}/users
        // { "username": "...", "enabled": true, "credentials": [...] }
        // throws UsernameConflictException if 409
        // throws RegistrationException for other errors
    }
}

Important: this client logs in to the master realm with admin credentials to get an admin token, then uses that token to manage users in the mapreduce realm. The admin credentials come from env vars.

---

6. manager-service — The Brain

Port: 8081 (ClusterIP — not exposed externally, only reachable from inside the cluster)

Role: This service does the heavy lifting. It owns the database, talks to Kubernetes, reads/writes MinIO, and orchestrates the entire MapReduce job lifecycle.

6.1 Dependencies

spring-boot-starter-web                  → internal REST API
spring-boot-starter-security             → validate JWT on requests from UI
spring-boot-starter-oauth2-resource-server → JWT decoding via Keycloak JWKS
spring-boot-starter-data-jpa            → ORM with Hibernate
postgresql                              → JDBC driver
flyway-core + flyway-database-postgresql → SQL migrations on startup
spring-boot-starter-validation          → @Valid on request bodies
spring-boot-starter-actuator            → health probes
micrometer-registry-prometheus          → metrics
io.fabric8:kubernetes-client (6.13.1)   → create/delete K8s Jobs at runtime
io.minio:minio (8.5.11)                 → read/write MinIO object storage
lombok                                  → code generation
mapstruct                               → entity ↔ DTO mapping
common (internal)                       → shared DTOs and enums

6.2 Configuration (application.yml)

server:
  port: 8081

spring:
  datasource:
    url: jdbc:postgresql://${DB_HOST:localhost}:${DB_PORT:5432}/${DB_NAME:mapreduce}
    username: ${DB_USER}
    password: ${DB_PASSWORD}
  jpa:
    hibernate:
      ddl-auto: validate       # Flyway manages schema, Hibernate just validates
    show-sql: false
  flyway:
    enabled: true
    locations: classpath:db/migration
  security:
    oauth2:
      resourceserver:
        jwt:
          issuer-uri: ${KEYCLOAK_ISSUER_URI}
          jwk-set-uri: ${KEYCLOAK_JWK_SET_URI}
  servlet:
    multipart:
      max-file-size: 1GB

minio:
  endpoint:        ${MINIO_ENDPOINT:http://minio:9000}
  public-endpoint: ${MINIO_PUBLIC_ENDPOINT:http://localhost:9000}   # for presigned URLs
  access-key:      ${MINIO_ACCESS_KEY}
  secret-key:      ${MINIO_SECRET_KEY}
  bucket:          ${MINIO_BUCKET:mapreduce}

k8s:
  namespace:   ${K8S_NAMESPACE:mapreduce}
  worker:
    image:         ${WORKER_IMAGE:mapreduce/worker:latest}
    backoff-limit: ${K8S_BACKOFF_LIMIT:3}

app:
  manager:
    internal-url: ${MANAGER_INTERNAL_URL:http://manager-service:8081}

mapreduce:
  default-map-tasks: ${DEFAULT_MAP_TASKS:4}
  watchdog:
    interval-ms: 15000

Two MinIO endpoint configs exist because:

• minio.endpoint is used for actual data operations (internal cluster DNS, fast)
• minio.public-endpoint is used when generating presigned URLs that the CLI/browser will download from (has to be a URL the client can reach)

6.3 Security Configuration

The manager has a two-tier security config using Spring Security's @Order on filter chains:

@Configuration
@EnableWebSecurity
public class SecurityConfig {

    // Chain 1 (priority order=1): worker callbacks and health checks → no auth
    @Bean @Order(1)
    public SecurityFilterChain workerCallbackChain(HttpSecurity http) throws Exception {
        return http
            .securityMatcher("/internal/tasks/**", "/actuator/health/**", "/actuator/prometheus")
            .csrf(csrf -> csrf.disable())
            .sessionManagement(s -> s.sessionCreationPolicy(STATELESS))
            .authorizeHttpRequests(auth -> auth.anyRequest().permitAll())
            .build();
    }

    // Chain 2 (lower priority): everything else → JWT required
    @Bean @Order(2)
    public SecurityFilterChain jwtFilterChain(HttpSecurity http) throws Exception {
        return http
            .csrf(csrf -> csrf.disable())
            .sessionManagement(s -> s.sessionCreationPolicy(STATELESS))
            .authorizeHttpRequests(auth -> auth.anyRequest().authenticated())
            .oauth2ResourceServer(oauth2 -> oauth2.jwt(jwt -> jwt
                .decoder(jwtDecoder())
                .jwtAuthenticationConverter(keycloakJwtConverter())
            ))
            .build();
    }
}

Why do workers get a free pass? Because worker pods are created by the manager dynamically — they don't have a JWT. They only know the manager URL and their task ID. The internal task endpoints don't need JWT because they're on a ClusterIP service (not reachable from outside the cluster) and the worker already needs to know the task ID to call them.

6.4 Entity Classes

Entities are JPA-managed objects that map to database tables.

Job entity:

@Entity @Table(name = "jobs")
@Getter @Setter @NoArgsConstructor
public class Job {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID jobId;

    @Column(name = "user_id", nullable = false)
    private String userId;          // Keycloak subject claim (user UUID)

    @Enumerated(EnumType.STRING)
    private JobStatus status;       // stored as VARCHAR "MAP_PHASE" etc.

    private String codePath;        // MinIO key: users/{userId}/code/{uuid}_name.jar
    private String inputPath;       // MinIO prefix: users/{userId}/raw/
    private String outputPath;      // MinIO prefix: users/{userId}/results/{jobId}/

    private int numMapTasks;
    private int numReduceTasks;

    private String errorMessage;

    @CreationTimestamp
    private Instant createdAt;

    @UpdateTimestamp
    private Instant updatedAt;

    @OneToMany(mappedBy = "job", cascade = CascadeType.ALL, fetch = FetchType.LAZY)
    private List<Task> tasks = new ArrayList<>();
}

Task entity:

@Entity @Table(name = "tasks")
@Getter @Setter @NoArgsConstructor
public class Task {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID taskId;

    @ManyToOne(fetch = FetchType.LAZY)
    @JoinColumn(name = "job_id")
    private Job job;

    @Enumerated(EnumType.STRING)
    private TaskType taskType;       // MAP or REDUCE

    @Enumerated(EnumType.STRING)
    private TaskStatus status;       // IDLE, IN_PROGRESS, COMPLETED, FAILED

    private String workerPodId;      // K8s Job name: worker-map-{taskId}
    private String inputSplit;       // comma-separated MinIO object keys
    private String outputLocation;   // MinIO path where worker wrote output

    private int retryCount;          // incremented on each failure, max 3

    private Instant lastHeartbeat;   // updated every 10s by the worker

    private String errorMessage;

    @CreationTimestamp private Instant createdAt;
    @UpdateTimestamp  private Instant updatedAt;
}

FileMetadata entity:

@Entity @Table(name = "file_metadata")
@Getter @Setter @NoArgsConstructor
public class FileMetadata {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID fileId;

    private String userId;         // Keycloak sub
    private String fileType;       // "DATA" or "CODE"
    private String originalName;   // original filename from upload
    private String storagePath;    // MinIO object key

    private Long sizeBytes;

    @CreationTimestamp
    private Instant createdAt;
}

6.5 Repositories

Spring Data JPA repositories — just interfaces, Spring generates the implementation.

public interface JobRepository extends JpaRepository<Job, UUID> {
    List<Job> findByUserId(String userId);
    List<Job> findByStatus(JobStatus status);
    Optional<Job> findByJobIdAndUserId(UUID jobId, String userId);
}

public interface TaskRepository extends JpaRepository<Task, UUID> {
    List<Task> findByJobJobId(UUID jobId);
    List<Task> findByJobJobIdAndTaskType(UUID jobId, TaskType type);
    List<Task> findByJobJobIdAndStatus(UUID jobId, TaskStatus status);
    List<Task> findByStatusAndLastHeartbeatBefore(TaskStatus status, Instant threshold);
    long countByJobJobIdAndStatus(UUID jobId, TaskStatus status);
    long countByJobJobIdAndTaskType(UUID jobId, TaskType type);
}

public interface FileMetadataRepository extends JpaRepository<FileMetadata, UUID> {
    Optional<FileMetadata> findByFileIdAndUserId(UUID fileId, String userId);
    List<FileMetadata> findByUserIdAndFileType(String userId, String fileType);
}

The findByStatusAndLastHeartbeatBefore query in TaskRepository is specifically used by the heartbeat watchdog to find stuck tasks.

6.6 JobOrchestrationService — The Most Important Class

This is the brain of the whole system. Every job lifecycle event flows through here.

@Service @RequiredArgsConstructor @Slf4j
public class JobOrchestrationService {

    private final JobRepository jobRepository;
    private final TaskRepository taskRepository;
    private final FileMetadataRepository fileMetadataRepository;
    private final KubernetesJobLauncher k8sLauncher;
    private final MinioStorageService minioService;

    @Value("${mapreduce.default-map-tasks:4}")
    private int defaultMapTasks;

submitJob(JobSubmitRequest request, String userId) — called when a user submits a job:

1. Looks up FileMetadata for dataId and codeId (checks they belong to this user)
2. Creates a Job entity with status INITIALIZING
3. Sets outputPath = users/{userId}/results/{jobId}/
4. Saves to DB
5. Returns the jobId
6. Asynchronously (via @Async or just in a new thread) calls startMapPhase(job)

startMapPhase(Job job):

1. Lists all MinIO objects under job.getInputPath() (the user's uploaded data files)
2. Splits the list into numMapTasks chunks (round-robin assignment)
3. For each chunk:
   • Creates a Task entity (taskType=MAP, status=IDLE, inputSplit=key1,key2,...)
   • Saves to DB
4. Updates job status to MAP_PHASE
5. For each task, calls k8sLauncher.launchWorker(...) and stores the returned K8s job name as workerPodId
6. Updates all tasks from IDLE to IN_PROGRESS (well, the workers will also confirm this themselves)

handleTaskUpdate(UUID taskId, TaskStatusUpdate update) — called when a worker reports back:

@Transactional
public void handleTaskUpdate(UUID taskId, TaskStatusUpdate update) {
    Task task = taskRepository.findById(taskId).orElseThrow();
    Job job = task.getJob();

    switch (update.getStatus()) {
        case IN_PROGRESS -> {
            task.setStatus(TaskStatus.IN_PROGRESS);
            task.setLastHeartbeat(Instant.now());
        }
        case COMPLETED -> {
            task.setStatus(TaskStatus.COMPLETED);
            task.setOutputLocation(update.getOutputLocation());
            onTaskCompleted(job, task);
        }
        case FAILED -> {
            task.setStatus(TaskStatus.FAILED);
            task.setErrorMessage(update.getErrorMessage());
            onTaskFailed(job, task);
        }
    }
    taskRepository.save(task);
}

onTaskCompleted(Job job, Task completedTask):

• Counts how many tasks of the current type are completed
• If this is a MAP task and ALL maps are now done → calls startReducePhase(job)
• If this is a REDUCE task and ALL reduces are done → sets job to COMPLETED, generates output URLs

startReducePhase(Job job):

1. Collects outputLocation from all completed MAP tasks
2. Splits those MinIO paths across numReduceTasks reducers (round-robin)
3. Creates Task entities (taskType=REDUCE, inputSplit = paths from map outputs)
4. Updates job status to REDUCE_PHASE
5. Launches a K8s worker Job for each reduce task

onTaskFailed(Job job, Task failedTask):

private void onTaskFailed(Job job, Task failedTask) {
    if (failedTask.getRetryCount() < 3) {
        failedTask.setRetryCount(failedTask.getRetryCount() + 1);
        failedTask.setStatus(TaskStatus.IDLE);
        // delete old K8s job (if it still exists)
        if (failedTask.getWorkerPodId() != null) {
            k8sLauncher.deleteJob(failedTask.getWorkerPodId());
        }
        // launch a new worker for the same task
        String newPodId = k8sLauncher.launchWorker(
            failedTask.getTaskId(), job.getJobId(),
            failedTask.getTaskType().name(),
            failedTask.getInputSplit(), job.getCodePath()
        );
        failedTask.setWorkerPodId(newPodId);
    } else {
        // no more retries → job fails permanently
        job.setStatus(JobStatus.FAILED);
        job.setErrorMessage("Task " + failedTask.getTaskId() + " failed after 3 retries");
        jobRepository.save(job);
    }
}

cancelJob(UUID jobId, String userId):

1. Looks up job (checks ownership)
2. Sets status to CANCELLED
3. For each task that is not completed, deletes the corresponding K8s Job via k8sLauncher.deleteJob(task.getWorkerPodId())

getJobStatus(UUID jobId, String userId):

When the job is COMPLETED, this generates presigned download URLs:

1. Lists all MinIO objects under job.getOutputPath() (the results/ prefix)
2. For each object key, calls minioService.presignedGetUrl(key, 3600) (1 hour expiry)
3. Returns JobStatusResponse with those URLs in outputUrls

6.7 FileService

@Service @RequiredArgsConstructor @Slf4j
public class FileService {

    private final MinioStorageService minioService;
    private final FileMetadataRepository fileMetadataRepository;

    @Transactional
    public FileUploadResponse uploadData(MultipartFile file, String userId) {
        String key = "users/" + userId + "/raw/" + UUID.randomUUID() + "_" + file.getOriginalFilename();
        return upload(file, userId, "DATA", key);
    }

    @Transactional
    public FileUploadResponse uploadCode(MultipartFile file, String userId) {
        String key = "users/" + userId + "/code/" + UUID.randomUUID() + "_" + file.getOriginalFilename();
        return upload(file, userId, "CODE", key);
    }

    private FileUploadResponse upload(MultipartFile file, String userId, String type, String key) {
        minioService.upload(key, file.getInputStream(), file.getSize(), file.getContentType());

        FileMetadata meta = new FileMetadata();
        meta.setUserId(userId);
        meta.setFileType(type);
        meta.setOriginalName(file.getOriginalFilename());
        meta.setStoragePath(key);
        meta.setSizeBytes(file.getSize());
        fileMetadataRepository.save(meta);

        return FileUploadResponse.builder()
            .id(meta.getFileId().toString())
            .storagePath(key)
            .build();
    }
}

6.8 KubernetesJobLauncher

This is how the manager creates worker pods. It uses the Fabric8 Kubernetes client (Java SDK for the Kubernetes API).

@Service @RequiredArgsConstructor @Slf4j
public class KubernetesJobLauncher {

    private final KubernetesClient kubernetesClient;

    @Value("${k8s.namespace:mapreduce}")       private String namespace;
    @Value("${k8s.worker.image}")              private String workerImage;
    @Value("${k8s.worker.backoff-limit:3}")    private int backoffLimit;
    @Value("${app.manager.internal-url}")      private String managerUrl;
    @Value("${minio.endpoint}")                private String minioEndpoint;
    @Value("${minio.access-key}")              private String minioAccessKey;
    @Value("${minio.secret-key}")              private String minioSecretKey;
    @Value("${minio.bucket}")                  private String minioBucket;

launchWorker(UUID taskId, UUID jobId, String taskType, String inputSplit, String codePath):

This method builds a full batch/v1/Job spec using Fabric8's fluent builder API:

Job k8sJob = new JobBuilder()
    .withNewMetadata()
        .withName("worker-" + taskType.toLowerCase() + "-" + taskId)
        .withNamespace(namespace)
        .addToLabels("app", "mapreduce-worker")
        .addToLabels("job-id", jobId.toString())
        .addToLabels("task-id", taskId.toString())
        .addToLabels("task-type", taskType)
    .endMetadata()
    .withNewSpec()
        .withBackoffLimit(backoffLimit)   // K8s-level retry limit
        .withNewTemplate()
            .withNewSpec()
                .withRestartPolicy("Never")   // don't restart the pod, K8s creates new one
                .addNewContainer()
                    .withName("worker")
                    .withImage(workerImage)
                    .withEnv(buildEnvVars(taskId, jobId, taskType, inputSplit, codePath))
                    .withNewResources()
                        .addToRequests("cpu",    new Quantity("250m"))
                        .addToRequests("memory", new Quantity("256Mi"))
                        .addToLimits("cpu",      new Quantity("1"))
                        .addToLimits("memory",   new Quantity("512Mi"))
                    .endResources()
                .endContainer()
            .endSpec()
        .endTemplate()
    .endSpec()
    .build();

kubernetesClient.batch().v1().jobs()
    .inNamespace(namespace)
    .resource(k8sJob)
    .create();

The buildEnvVars(...) method creates a list of EnvVar objects for all the environment variables the worker needs:

TASK_ID           → taskId.toString()
JOB_ID            → jobId.toString()
TASK_TYPE         → "MAP" or "REDUCE"
INPUT_SPLIT       → "key1,key2,key3"
CODE_PATH         → "users/{userId}/code/{uuid}_code.jar"
MANAGER_URL       → "http://manager-service:8081"
MINIO_ENDPOINT    → "http://minio:9000"
MINIO_ACCESS_KEY  → from manager's config
MINIO_SECRET_KEY  → from manager's config
MINIO_BUCKET      → "mapreduce"

deleteJob(String k8sJobName):

public void deleteJob(String k8sJobName) {
    kubernetesClient.batch().v1().jobs()
        .inNamespace(namespace)
        .withName(k8sJobName)
        .withPropagationPolicy(DeletionPropagation.BACKGROUND)  // also deletes pods
        .delete();
}

6.9 MinioStorageService

@Service @Slf4j
public class MinioStorageService {

    private final MinioClient minioClient;         // internal endpoint (@Primary)
    private final MinioClient publicMinioClient;   // public endpoint (@Qualifier)

    @Value("${minio.bucket}")
    private String bucket;

    public String upload(String objectKey, InputStream data, long size, String contentType) {
        ensureBucketExists();
        minioClient.putObject(PutObjectArgs.builder()
            .bucket(bucket).object(objectKey)
            .stream(data, size, -1)
            .contentType(contentType)
            .build());
        return objectKey;
    }

    public InputStream download(String objectKey) {
        return minioClient.getObject(GetObjectArgs.builder()
            .bucket(bucket).object(objectKey).build());
    }

    public List<String> listObjects(String prefix) {
        List<String> keys = new ArrayList<>();
        Iterable<Result<Item>> results = minioClient.listObjects(
            ListObjectsArgs.builder().bucket(bucket).prefix(prefix).recursive(true).build()
        );
        for (Result<Item> result : results) {
            keys.add(result.get().objectName());
        }
        return keys;
    }

    // Uses the PUBLIC endpoint so the presigned URL is accessible from outside the cluster
    public String presignedGetUrl(String objectKey, int expirySeconds) {
        return publicMinioClient.getPresignedObjectUrl(GetPresignedObjectUrlArgs.builder()
            .bucket(bucket).object(objectKey)
            .method(Method.GET)
            .expiry(expirySeconds, TimeUnit.SECONDS)
            .build());
    }
}

The reason there are two MinioClient beans (MinioConfig.java): when the manager is running inside the cluster, it accesses MinIO via http://minio:9000 (cluster DNS). But presigned URLs need to use the public address (like http://192.168.49.2:30900) because the CLI runs on the user's laptop — it can't resolve minio as a hostname.

6.10 HeartbeatWatchdog

@Component @RequiredArgsConstructor @Slf4j
public class HeartbeatWatchdog {

    private final TaskRepository taskRepository;
    private final JobOrchestrationService orchestrationService;

    private static final long TIMEOUT_SECONDS = 30;

    @Scheduled(fixedDelayString = "${mapreduce.watchdog.interval-ms:15000}")
    public void checkDeadWorkers() {
        Instant threshold = Instant.now().minusSeconds(TIMEOUT_SECONDS);
        List<Task> deadTasks = taskRepository
            .findByStatusAndLastHeartbeatBefore(TaskStatus.IN_PROGRESS, threshold);

        for (Task task : deadTasks) {
            log.warn("Task {} has no heartbeat since {} — marking as FAILED", task.getTaskId(), threshold);
            TaskStatusUpdate update = new TaskStatusUpdate();
            update.setStatus(TaskStatus.FAILED);
            update.setErrorMessage("Heartbeat timeout — worker pod likely died");
            orchestrationService.handleTaskUpdate(task.getTaskId(), update);
        }
    }
}

This runs every 15 seconds. If a task is IN_PROGRESS but hasn't sent a heartbeat in 30 seconds, the watchdog marks it as FAILED. This triggers onTaskFailed which either retries the task or fails the whole job.

The @EnableScheduling annotation is on ManagerApplication — without it, @Scheduled annotations are ignored.

6.11 Internal Controllers

InternalJobController — /internal/jobs:

@GetMapping                         → listJobs(@AuthenticationPrincipal Jwt jwt)
@PostMapping                        → submitJob(@Valid @RequestBody, Jwt jwt)
@GetMapping("/{jobId}")             → getJobStatus(@PathVariable UUID, Jwt jwt)
@DeleteMapping("/{jobId}")          → cancelJob(@PathVariable UUID, Jwt jwt)

The @AuthenticationPrincipal Jwt jwt parameter gets the decoded JWT. The user's ID is extracted as jwt.getSubject() (the sub claim), which is Keycloak's UUID for that user. This becomes userId in all DB operations.

InternalTaskController — /internal/tasks (no auth required):

@PostMapping("/{taskId}/status")    → updateTaskStatus(@PathVariable UUID, @Valid body)
@PostMapping("/{taskId}/heartbeat") → heartbeat(@PathVariable UUID)

InternalFileController — /internal/files:

@PostMapping("/data")   → uploadData(@RequestParam("file") MultipartFile, Jwt jwt)
@PostMapping("/code")   → uploadCode(@RequestParam("file") MultipartFile, Jwt jwt)

---

7. worker — The Compute Pod

The worker is fundamentally different from the other services. It's a Spring Boot CommandLineRunner — meaning it starts up, runs one task, and exits. It's not a server that accepts requests. It communicates by calling the manager's REST API.

7.1 WorkerApplication

@SpringBootApplication
@RequiredArgsConstructor
@Slf4j
public class WorkerApplication implements CommandLineRunner {

    private final WorkerRunner workerRunner;

    public static void main(String[] args) {
        SpringApplication.run(WorkerApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        workerRunner.execute();
    }
}

The application.yml disables a bunch of Spring Boot auto-configuration that would try to connect to a database or set up security — the worker doesn't need any of that:

spring:
  autoconfigure:
    exclude:
      - org.springframework.boot.autoconfigure.security.SecurityAutoConfiguration
      - org.springframework.boot.autoconfigure.security.servlet.UserDetailsServiceAutoConfiguration
      - org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
      - org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration

server:
  port: 0    # no HTTP server needed — worker only makes outbound calls

worker:
  task-id:     ${TASK_ID}
  job-id:      ${JOB_ID}
  task-type:   ${TASK_TYPE}
  input-split: ${INPUT_SPLIT}
  code-path:   ${CODE_PATH}
  manager-url: ${MANAGER_URL}

minio:
  endpoint:   ${MINIO_ENDPOINT}
  access-key: ${MINIO_ACCESS_KEY}
  secret-key: ${MINIO_SECRET_KEY}
  bucket:     ${MINIO_BUCKET}

All the ${ENV_VAR} values come from the Kubernetes Job spec built by KubernetesJobLauncher.

7.2 WorkerRunner

@Component @RequiredArgsConstructor @Slf4j
public class WorkerRunner {

    private final WorkerMinioClient minioClient;
    private final ManagerReporter reporter;

    @Value("${worker.task-type}")   private String taskType;
    @Value("${worker.task-id}")     private String taskId;
    @Value("${worker.job-id}")      private String jobId;
    @Value("${worker.input-split}") private String inputSplit;   // comma-separated
    @Value("${worker.code-path}")   private String codePath;
    @Value("${minio.bucket}")       private String bucket;

    private static final long HEARTBEAT_INTERVAL_MS = 10_000;

execute() — the main method:

public void execute() {
    // 1. Tell the manager we're starting
    reporter.reportInProgress();

    // 2. Start sending heartbeats in background thread
    ScheduledExecutorService heartbeat = Executors.newSingleThreadScheduledExecutor();
    heartbeat.scheduleAtFixedRate(
        reporter::sendHeartbeat,
        HEARTBEAT_INTERVAL_MS, HEARTBEAT_INTERVAL_MS, TimeUnit.MILLISECONDS
    );

    try {
        // 3. Create a temp directory for working files
        Path workDir = Files.createTempDirectory("worker-" + taskId);

        // 4. Download the user's code JAR from MinIO to /tmp/worker-{taskId}/code.jar
        Path jarPath = workDir.resolve("code.jar");
        minioClient.downloadToFile(codePath, jarPath.toString());

        // 5. Run the appropriate phase
        String outputLocation;
        if ("MAP".equals(taskType)) {
            outputLocation = runMap(workDir, jarPath);
        } else {
            outputLocation = runReduce(workDir, jarPath);
        }

        // 6. Report success
        reporter.reportCompleted(outputLocation);

    } catch (Exception e) {
        log.error("Task {} failed: {}", taskId, e.getMessage(), e);
        reporter.reportFailed(e.getMessage());
    } finally {
        heartbeat.shutdownNow();
    }
}

runMap(Path workDir, Path jarPath) — step by step:

private String runMap(Path workDir, Path jarPath) throws Exception {
    // 1. Load the user's Mapper class from their JAR using ServiceLoader
    URLClassLoader classLoader = new URLClassLoader(
        new URL[]{jarPath.toUri().toURL()},
        Thread.currentThread().getContextClassLoader()
    );
    ServiceLoader<Mapper> loader = ServiceLoader.load(Mapper.class, classLoader);
    Mapper mapper = loader.findFirst()
        .orElseThrow(() -> new RuntimeException("No Mapper implementation found in JAR"));

    // 2. Process each input file in the split
    TreeMap<String, List<String>> intermediate = new TreeMap<>();
    for (String objectKey : inputSplit.split(",")) {
        String content = minioClient.downloadAsString(objectKey.trim());
        String[] lines = content.split("\n");
        for (int i = 0; i < lines.length; i++) {
            // key is "filename:lineNumber", value is the line text
            mapper.map(objectKey + ":" + i, lines[i], (k, v) -> {
                intermediate.computeIfAbsent(k, x -> new ArrayList<>()).add(v);
            });
        }
    }

    // 3. Write intermediate results: "key\tvalue" per line
    // One output file: temp/{jobId}/map-{taskId}/part-0.txt
    StringBuilder sb = new StringBuilder();
    for (Map.Entry<String, List<String>> entry : intermediate.entrySet()) {
        for (String val : entry.getValue()) {
            sb.append(entry.getKey()).append('\t').append(val).append('\n');
        }
    }
    String outputKey = "temp/" + jobId + "/map-" + taskId + "/part-0.txt";
    minioClient.uploadText(outputKey, sb.toString());

    return outputKey;   // returned as outputLocation in the status report
}

runReduce(Path workDir, Path jarPath) — step by step:

private String runReduce(Path workDir, Path jarPath) throws Exception {
    // 1. Load the user's Reducer from their JAR
    URLClassLoader classLoader = new URLClassLoader(
        new URL[]{jarPath.toUri().toURL()},
        Thread.currentThread().getContextClassLoader()
    );
    ServiceLoader<Reducer> loader = ServiceLoader.load(Reducer.class, classLoader);
    Reducer reducer = loader.findFirst()
        .orElseThrow(() -> new RuntimeException("No Reducer implementation found in JAR"));

    // 2. Download all assigned intermediate files and group by key
    TreeMap<String, List<String>> grouped = new TreeMap<>();
    for (String objectKey : inputSplit.split(",")) {
        String content = minioClient.downloadAsString(objectKey.trim());
        for (String line : content.split("\n")) {
            if (line.isBlank()) continue;
            int tab = line.indexOf('\t');
            String key = line.substring(0, tab);
            String val = line.substring(tab + 1);
            grouped.computeIfAbsent(key, x -> new ArrayList<>()).add(val);
        }
    }

    // 3. Run reduce for each key
    StringBuilder result = new StringBuilder();
    for (Map.Entry<String, List<String>> entry : grouped.entrySet()) {
        reducer.reduce(entry.getKey(), entry.getValue(), (k, v) -> {
            result.append(k).append('\t').append(v).append('\n');
        });
    }

    // 4. Write final output
    // Extract userId from codePath convention: users/{userId}/code/...
    String userId = extractUserId();
    String outputKey = "users/" + userId + "/results/" + jobId + "/reduce-" + taskId + ".txt";
    minioClient.uploadText(outputKey, result.toString());

    return outputKey;
}

The ServiceLoader mechanism — how user code gets loaded:

Java's ServiceLoader is a standard way to discover implementations of an interface at runtime. For it to work, the user's JAR must contain a file at:

META-INF/services/gr.tuc.distributed.common.api.Mapper

...containing exactly one line: the fully-qualified class name of their Mapper implementation. Same for Reducer. When the worker calls ServiceLoader.load(Mapper.class, classLoader), Java finds that file in the JAR and instantiates the class.

The URLClassLoader is key here — it creates a class loader that knows about the user's JAR file so it can load classes from it.

7.3 WorkerMinioClient

@Component @Slf4j
public class WorkerMinioClient {

    // builds MinioClient from @Value properties in constructor
    @Value("${minio.endpoint}")   private String endpoint;
    @Value("${minio.access-key}") private String accessKey;
    @Value("${minio.secret-key}") private String secretKey;
    @Value("${minio.bucket}")     private String bucket;

    public String downloadAsString(String objectKey)              { ... }  // GetObject → UTF-8 string
    public void   downloadToFile(String objectKey, String path)   { ... }  // download to disk
    public String uploadText(String objectKey, String content)    { ... }  // PutObject from string
    public String uploadFile(String objectKey, String path, String ct) { ... } // upload from disk
}

7.4 ManagerReporter

@Component @Slf4j
public class ManagerReporter {

    @Value("${worker.manager-url}") private String managerUrl;
    @Value("${worker.task-id}")     private String taskId;

    private final RestClient restClient;   // configured with managerUrl base

    public void reportInProgress() {
        post(TaskStatusUpdate with status=IN_PROGRESS);
    }

    public void reportCompleted(String outputLocation) {
        post(TaskStatusUpdate with status=COMPLETED, outputLocation=outputLocation);
    }

    public void reportFailed(String errorMessage) {
        post(TaskStatusUpdate with status=FAILED, errorMessage=errorMessage);
    }

    public void sendHeartbeat() {
        restClient.post()
            .uri("/internal/tasks/" + taskId + "/heartbeat")
            .retrieve().toBodilessEntity();
    }

    private void post(TaskStatusUpdate update) {
        restClient.post()
            .uri("/internal/tasks/" + taskId + "/status")
            .contentType(MediaType.APPLICATION_JSON)
            .body(update)
            .retrieve().toBodilessEntity();
    }
}

---

8. cli — The Command-Line Tool

The CLI is a fat JAR built with Maven Shade plugin. It's a plain Java 21 app — no Spring Boot, no DI framework, just picocli + Jackson.

8.1 Entry Point and Shell Mode (Main.java)

public class Main {
    public static void main(String[] args) {
        if (args.length == 0) {
            // No arguments → drop into interactive REPL shell
            runShell();
        } else {
            // Arguments given → run that command and exit
            int code = new CommandLine(new MapReduceCli()).execute(args);
            System.exit(code);
        }
    }

    private static void runShell() {
        CommandLine cmd = new CommandLine(new MapReduceCli());
        Console console = System.console();

        Out.bold("MapReduce CLI — type 'help' or 'exit'");
        Out.println("");

        while (true) {
            String line = console != null
                ? console.readLine("mr> ")
                : new Scanner(System.in).nextLine();

            if (line == null || line.trim().equalsIgnoreCase("exit")
                             || line.trim().equalsIgnoreCase("quit")) {
                break;
            }
            if (line.isBlank()) continue;

            // tokenize the line respecting quoted strings
            String[] tokens = tokenize(line.trim());
            cmd.execute(tokens);
        }
    }
}

The shell tokenizer handles quoted arguments — so mr run "my file.txt" code.jar works correctly even with spaces in filenames.

8.2 Root Command (MapReduceCli.java)

@Command(
    name = "mr",
    version = "1.0",
    mixinStandardHelpOptions = true,
    subcommands = {
        LoginCommand.class,
        LogoutCommand.class,
        RegisterCommand.class,
        WhoamiCommand.class,
        UploadCommand.class,
        SubmitCommand.class,
        RunCommand.class,
        JobsCommand.class,
        StatusCommand.class,
        CancelCommand.class,
        ResultsCommand.class
    }
)
public class MapReduceCli implements Callable<Integer> {
    @Override
    public Integer call() {
        // no subcommand given → print help
        new CommandLine(this).usage(System.out);
        return 0;
    }
}

8.3 Session Management

Session.java — the data stored in ~/.mr/session.json:

@JsonIgnoreProperties(ignoreUnknown = true)
public class Session {
    public String baseUrl;       // http://192.168.49.2:30080
    public String token;         // JWT access token (eyJ...)
    public String username;      // alice
    public long expiresAt;       // System.currentTimeMillis() + expires_in * 1000
    public String lastDataId;    // UUID of last uploaded data file
    public String lastCodeId;    // UUID of last uploaded code JAR

    public boolean isExpired() {
        return System.currentTimeMillis() > expiresAt;
    }

    public String remainingTime() {
        long remaining = (expiresAt - System.currentTimeMillis()) / 1000;
        return String.format("%02d:%02d", remaining / 60, remaining % 60);
    }
}

SessionStore.java — loads and saves the session file:

public class SessionStore {
    private static final Path SESSION_PATH =
        Path.of(System.getProperty("user.home"), ".mr", "session.json");

    private static final ObjectMapper MAPPER = new ObjectMapper();

    public static Optional<Session> load() { ... }
    public static void save(Session s) { ... }
    public static void clear() { ... }

    // Used by commands that require login — throws instead of System.exit
    // so the interactive shell doesn't die when you forget to log in
    public static Session require() {
        return load()
            .filter(s -> !s.isExpired())
            .orElseThrow(() -> new IllegalStateException(
                "Not logged in. Run 'mr login' first."
            ));
    }
}

8.4 ApiClient.java

Pure Java, uses java.net.http.HttpClient (built into Java 11+). No third-party HTTP library needed.

public class ApiClient {
    private static final HttpClient HTTP = HttpClient.newBuilder()
        .connectTimeout(Duration.ofSeconds(10))
        .build();

    private static final ObjectMapper MAPPER = new ObjectMapper()
        .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

    // Response records (Java 16+ records)
    public record LoginResponse(String access_token, long expires_in) {}
    public record FileUploadResponse(String id, String fileId) {
        public String resolvedId() { return id != null ? id : fileId; }
    }
    public record JobStatus(
        String jobId, String status, String errorMessage,
        List<String> outputUrls, String createdAt, String updatedAt
    ) {}

The MultipartBodyPublisher utility builds the multipart body for file uploads:

// cli/util/MultipartBodyPublisher.java
// Builds application/x-www-form-urlencoded / multipart/form-data bodies
// for java.net.http.HttpRequest — the standard library doesn't have a built-in multipart builder

8.5 Command Classes

LoginCommand

@Command(name = "login", description = "Log in to the MapReduce cluster.")
public class LoginCommand implements Callable<Integer> {

    @Option(names={"--url","-u"}, defaultValue="http://192.168.49.2:30080")
    String url;

    @Option(names={"--username","-n"}, description="Username (prompted if omitted)")
    String username;

    @Option(names={"--password","-p"}, description="Password (prompted securely if omitted)", interactive=true)
    char[] password;

    @Override
    public Integer call() {
        // prompts for username/password if not supplied via flags
        // calls api.login() → gets JWT
        // saves Session to ~/.mr/session.json with expiresAt
        // prints "Logged in as alice (token expires in 04:59)"
    }
}

The interactive=true on the --password option means picocli will read it securely (no echo) from the console if not supplied on the command line.

RegisterCommand

@Command(name = "register")
public class RegisterCommand implements Callable<Integer> {

    @Option(names={"--url","-u"}, defaultValue="http://192.168.49.2:30080")
    String url;

    @Override
    public Integer call() {
        // 1. Prompt for username (min 3 chars)
        // 2. Check availability: api.isUsernameAvailable() → loop if taken
        // 3. Prompt for password (min 6 chars) with confirmation
        // 4. api.register()
        // 5. Print "Account created. Use 'mr login' to log in."
    }
}

UploadCommand

@Command(name = "upload", subcommands = {DataSubcommand.class, CodeSubcommand.class})
public class UploadCommand { }

@Command(name = "data")
static class DataSubcommand implements Callable<Integer> {
    @Parameters(index="0", paramLabel="<file>") Path file;

    @Override
    public Integer call() {
        Session s = SessionStore.require();
        var response = new ApiClient().uploadData(s.baseUrl, s.token, file);
        s.lastDataId = response.resolvedId();
        SessionStore.save(s);   // remember for mr submit
        Out.ok("Uploaded: " + response.resolvedId());
        return 0;
    }
}

Same pattern for CodeSubcommand with uploadCode and lastCodeId.

RunCommand

@Command(name = "run", description = "Upload data + code, submit job, watch status.")
public class RunCommand implements Callable<Integer> {

    @Parameters(index="0", paramLabel="<data>")    Path dataFile;
    @Parameters(index="1", paramLabel="<code.jar>") Path codeFile;

    @Option(names={"--reducers","-r"}, defaultValue="2")
    int reducers;

    @Option(names={"--watch","-w"}, defaultValue="5",
            description="Poll interval in seconds (0 to skip watching)")
    int watchInterval;

    @Override
    public Integer call() {
        Session s = SessionStore.require();
        var api = new ApiClient();

        // upload data
        Out.print("Uploading data...  ");
        var dataResp = api.uploadData(s.baseUrl, s.token, dataFile);
        Out.ok(dataResp.resolvedId());

        // upload code
        Out.print("Uploading code...  ");
        var codeResp = api.uploadCode(s.baseUrl, s.token, codeFile);
        Out.ok(codeResp.resolvedId());

        // submit job
        String jobId = api.submitJob(s.baseUrl, s.token,
            dataResp.resolvedId(), codeResp.resolvedId(), reducers);
        Out.ok("Job submitted: " + jobId);

        // watch until done
        if (watchInterval > 0) {
            while (true) {
                Thread.sleep(watchInterval * 1000L);
                var status = api.getJobStatus(s.baseUrl, s.token, jobId);
                Out.print("\rStatus: " + status.status() + "   ");  // \r overwrites line
                if (isTerminal(status.status())) {
                    Out.println("");
                    break;
                }
            }
        }
        return 0;
    }
}

The \r (carriage return) trick makes the status update overwrite the same line in the terminal — like a progress indicator.

ResultsCommand

@Command(name = "results", description = "View or download job output.")
public class ResultsCommand implements Callable<Integer> {

    @Parameters(index="0", paramLabel="<jobId>") String jobId;

    @Option(names={"--out","-o"}, description="Save raw output to file instead of displaying")
    Path outFile;

    @Option(names={"--top","-n"}, defaultValue="0", description="Show only top N results")
    int top;

The display logic:

1. Gets job status (must be COMPLETED)
2. Downloads all presigned URLs and concatenates the content
3. Parses each line as key\tvalue
4. Tries Long.parseLong(value) — if all values parse, treats as numeric
5. If numeric: sorts by count descending, shows table with proportional bar chart
6. --top N limits the display to the N highest-count entries
7. --out file skips all that and just writes raw content to disk

The bar chart is rendered with Unicode block characters via Out.bar(pct, 30) — generates a string of █ characters proportional to pct out of 100.

StatusCommand

@Command(name = "status")
public class StatusCommand implements Callable<Integer> {

    @Parameters(index="0", paramLabel="<jobId>") String jobId;

    @Option(names={"--watch","-w"}, defaultValue="0",
            description="Poll interval in seconds (0 = one-shot)")
    int watchInterval;

With --watch 5, it polls every 5 seconds and overwrites the status line until the job reaches a terminal state (COMPLETED, FAILED, or CANCELLED).

8.6 Out.java — Terminal Output Utilities

public class Out {
    // ANSI color codes for terminal
    private static final String RESET  = "\u001B[0m";
    private static final String GREEN  = "\u001B[32m";
    private static final String RED    = "\u001B[31m";
    private static final String YELLOW = "\u001B[33m";
    private static final String BOLD   = "\u001B[1m";
    private static final String DIM    = "\u001B[2m";

    public static void ok(String msg)   { System.out.println(GREEN + "✓ " + msg + RESET); }
    public static void err(String msg)  { System.err.println(RED   + "✗ " + msg + RESET); }
    public static void warn(String msg) { System.out.println(YELLOW+ "! " + msg + RESET); }
    public static void bold(String msg) { System.out.println(BOLD  + msg + RESET); }
    public static void dim(String msg)  { System.out.println(DIM   + msg + RESET); }
    public static void hr(int width)    { System.out.println("─".repeat(width)); }

    public static String bar(int pct, int width) {
        int filled = (int) Math.round(pct * width / 100.0);
        return "█".repeat(filled) + "░".repeat(width - filled);
    }
}

8.7 The mr Launcher Script

#!/bin/sh
DIR="$(cd "$(dirname "$0")" && pwd)"
JAR="$DIR/target/mr.jar"

if [ ! -f "$JAR" ]; then
    echo "mr.jar not found — building first..."
    cd "$DIR" && flatpak-spawn --host mvn package -q -DskipTests
fi

exec java -jar "$JAR" "$@"

flatpak-spawn --host is needed because VS Code runs as a Flatpak here, which sandboxes the shell — it can't see the host's Maven installation directly. flatpak-spawn --host breaks out of the sandbox and runs the command on the host OS.

---

9. examples/wordcount — Sample Job

This is how you write a MapReduce job for this system. It's deliberately simple.

public class WordCountMapper implements Mapper {
    private static final Pattern PUNCTUATION = Pattern.compile("[^a-zA-Z0-9']");
    private static final Pattern WHITESPACE  = Pattern.compile("\\s+");

    @Override
    public void map(String inputKey, String line, BiConsumer<String, String> emit) {
        if (line == null || line.isBlank()) return;
        for (String token : WHITESPACE.split(line)) {
            String word = PUNCTUATION.matcher(token.toLowerCase()).replaceAll("");
            if (!word.isBlank()) {
                emit.accept(word, "1");   // emit (word, "1") for each occurrence
            }
        }
    }
}

public class WordCountReducer implements Reducer {

    @Override
    public void reduce(String key, List<String> values, BiConsumer<String, String> emit) {
        long count = values.stream().mapToLong(Long::parseLong).sum();
        emit.accept(key, String.valueOf(count));
    }
}

ServiceLoader registration files:

# META-INF/services/gr.tuc.distributed.common.api.Mapper
gr.tuc.distributed.examples.wordcount.WordCountMapper

# META-INF/services/gr.tuc.distributed.common.api.Reducer
gr.tuc.distributed.examples.wordcount.WordCountReducer

The pom.xml for wordcount uses <scope>provided</scope> for the common dependency — meaning it's not bundled in the JAR. The worker already has common on its classpath, so there's no need to ship it twice. This keeps the JAR small.

---

10. External Infrastructure

10.1 PostgreSQL

• Image: postgres:16-alpine
• Purpose: Persistent state for jobs, tasks, and file metadata
• Who talks to it: Manager service only (via Spring Data JPA + Flyway)
• Database name: mapreduce
• Schema management: Flyway runs SQL migration files in order at startup. The naming convention matters: V{number}__{description}.sql. V1 runs before V2, etc.
• Connection details come from env vars: DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_PASSWORD

The manager uses a HikariCP connection pool (Spring Boot's default) — Hibernate holds a pool of database connections so it doesn't open a new TCP connection for every query.

10.2 MinIO

• Image: quay.io/minio/minio:latest
• What it is: An S3-compatible object store you run yourself
• Ports: 9000 (S3 API), 9001 (web console UI)
• Bucket: mapreduce (auto-created by MinioStorageService.ensureBucketExists() on first upload)
• Why MinIO instead of a real filesystem? Because multiple services need to access the same files. A local filesystem only works on one machine. MinIO gives you S3-compatible object storage that any service in the cluster can access via HTTP.

Storage path conventions:

users/{userId}/raw/{uuid}_{filename}           ← uploaded input data
users/{userId}/code/{uuid}_{filename}.jar      ← uploaded code JARs
temp/{jobId}/map-{taskId}/part-0.txt           ← MAP output (intermediate)
users/{userId}/results/{jobId}/reduce-{n}.txt  ← final REDUCE output

Presigned URLs: When you request results for a completed job, the manager generates presigned GET URLs for each output file. A presigned URL is a regular HTTPS URL with an embedded signature and expiry time — anyone who has it can download the file for up to 1 hour without needing credentials.

10.3 Keycloak

• Image: quay.io/keycloak/keycloak:25.0
• Mode: start-dev (no TLS, embedded H2 or PostgreSQL)
• Realm: mapreduce
• Client: mapreduce-cli (with Direct Access Grants enabled — needed for username/password login)

Keycloak issues JWT access tokens. The token contains:

• sub: user UUID (used as userId everywhere in the system)
• preferred_username: human-readable username
• realm_access.roles: list of roles (["default-roles-mapreduce", "offline_access"] for regular users, ["admin"] for admins)
• exp: expiry time
• iss: issuer URI (e.g. http://keycloak:8080/realms/mapreduce)

Both UI service and manager service validate tokens by fetching Keycloak's JWKS (JSON Web Key Set) endpoint at:

http://keycloak:8080/realms/mapreduce/protocol/openid-connect/certs

This endpoint returns the public keys. Spring Security caches these keys and uses them to verify the JWT signature without calling Keycloak on every request.

10.4 Kubernetes

• What it provides: Container orchestration — scheduling pods, managing resources, handling restarts
• Namespace: mapreduce — all resources for this system live here, isolated from other workloads
• Worker pods: Created dynamically at runtime by the manager service via the Fabric8 K8s client
• Why K8s for workers? Because you need to run many workers in parallel, each with specific env vars, resource limits, and restart policies. K8s batch/v1/Job is exactly designed for this use case — a task that runs to completion and exits.

---

11. Authentication Flows

11.1 Login

mr login --url http://host:30080

[User's machine]         [UI Service :30080]      [Keycloak :30180]
      │                         │                       │
      │  POST /api/auth/token   │                       │
      │  {username, password}   │                       │
      ├────────────────────────►│                       │
      │                         │  POST /realms/mapreduce
      │                         │  /openid-connect/token │
      │                         │  grant_type=password   │
      │                         │  client_id=mapreduce-cli
      │                         │  username=alice        │
      │                         │  password=secret123    │
      │                         ├──────────────────────►│
      │                         │  {                     │
      │                         │    access_token: eyJ...,│
      │                         │    expires_in: 300,    │
      │                         │    token_type: Bearer  │
      │                         │  }                     │
      │                         │◄──────────────────────┤
      │  {token, expiresAt,     │                       │
      │   username}             │                       │
      │◄────────────────────────┤                       │
      │                         │                       │
 [saves ~/.mr/session.json]

11.2 How JWT Validation Works on Every Request

CLI sends:  Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...

UI Service receives request:
  1. Spring Security intercepts it
  2. Extracts the Bearer token
  3. Calls NimbusJwtDecoder.decode(token):
     a. Splits JWT into header.payload.signature
     b. Gets public key from cached JWKS (fetches from Keycloak if not cached)
     c. Verifies signature using RS256 (RSA + SHA-256)
     d. Checks exp claim (token not expired)
     e. Checks iss claim matches configured issuer-uri
  4. If valid: creates Authentication object with sub, roles
  5. Puts Authentication in SecurityContextHolder
  6. Request proceeds to controller

UI Service then calls Manager:
  → Forwards the SAME Authorization: Bearer header

Manager receives request:
  → Does the same JWT validation (same Keycloak JWKS endpoint)
  → Extracts userId from jwt.getSubject()

11.3 Registration

mr register

[CLI]                       [UI Service]              [Keycloak Admin API]
 │                               │                          │
 │  POST /api/auth/register      │                          │
 │  {username, password}         │                          │
 ├──────────────────────────────►│                          │
 │                               │  GET /admin/realms/      │
 │                               │  mapreduce/users?        │
 │                               │  username=alice&exact=true│
 │                               │  (with admin token)      │
 │                               ├─────────────────────────►│
 │                               │  [] (empty = available)  │
 │                               │◄─────────────────────────┤
 │                               │  POST /admin/realms/     │
 │                               │  mapreduce/users         │
 │                               │  {username, enabled:true,│
 │                               │   credentials:[{type:    │
 │                               │   password, value:...}]} │
 │                               ├─────────────────────────►│
 │                               │  201 Created             │
 │                               │◄─────────────────────────┤
 │  200 OK                       │                          │
 │◄──────────────────────────────┤                          │

11.4 Logout

Logout is client-side only — the CLI just deletes ~/.mr/session.json. JWTs are stateless (the server doesn't keep a session table), so there's nothing to revoke server-side. The token will naturally expire based on the exp claim. In a production system you'd implement token revocation via an allowlist/denylist, but for this system, expiry is good enough.

---

12. File Upload Flow

mr upload data mybook.txt

[CLI]         [UI :30080]       [Manager :8081]     [MinIO :9000]   [PostgreSQL]
  │                │                   │                  │               │
  │ POST /api/v1/  │                   │                  │               │
  │ data           │                   │                  │               │
  │ multipart/form │                   │                  │               │
  │ Auth: Bearer   │                   │                  │               │
  ├───────────────►│                   │                  │               │
  │                │ validates JWT     │                  │               │
  │                │ extracts userId   │                  │               │
  │                │                   │                  │               │
  │                │ POST /internal/   │                  │               │
  │                │ files/data        │                  │               │
  │                │ (same multipart)  │                  │               │
  │                │ Auth: Bearer      │                  │               │
  │                ├──────────────────►│                  │               │
  │                │                   │ validates JWT    │               │
  │                │                   │                  │               │
  │                │                   │ key = users/     │               │
  │                │                   │ {userId}/raw/    │               │
  │                │                   │ {uuid}_mybook.txt│               │
  │                │                   │                  │               │
  │                │                   │ PutObject        │               │
  │                │                   ├─────────────────►│               │
  │                │                   │ 200 OK           │               │
  │                │                   │◄─────────────────┤               │
  │                │                   │                  │               │
  │                │                   │ INSERT file_meta │               │
  │                │                   │ {userId, "DATA", │               │
  │                │                   │  key, size}      │               │
  │                │                   ├───────────────────────────────────►
  │                │                   │ {fileId: uuid}   │               │
  │                │  {id: uuid}       │◄──────────────────────────────────┤
  │  {id: uuid}    │◄──────────────────┤                  │               │
  │◄───────────────┤                   │                  │               │

[CLI saves id as session.lastDataId]

---

13. Full Job Lifecycle

13.1 Submission

mr submit  (or mr run data.txt code.jar which does upload + submit)

[CLI]         [UI Service]     [Manager Service]     [PostgreSQL]
  │                │                   │                  │
  │ POST /api/v1/  │                   │                  │
  │ jobs           │                   │                  │
  │ {dataId,codeId,│                   │                  │
  │  numReducers}  │                   │                  │
  ├───────────────►│                   │                  │
  │                ├──────────────────►│                  │
  │                │                   │ looks up         │
  │                │                   │ file_metadata    │
  │                │                   │ for dataId       │
  │                │                   ├─────────────────►│
  │                │                   │                  │
  │                │                   │ INSERT jobs      │
  │                │                   │ status=INITIALIZING
  │                │                   ├─────────────────►│
  │                │                   │ {jobId: uuid}    │
  │  {jobId: uuid} │◄──────────────────┤                  │
  │◄───────────────┤                   │                  │
  │                │                   │                  │
  │                │               [async: startMapPhase]

The job submission is non-blocking — the HTTP request returns immediately with the jobId, then the map phase starts asynchronously. This is important so the CLI isn't hanging waiting for all workers to finish.

13.2 Map Phase — Step by Step

[Manager async]              [Kubernetes API]        [Worker Pod - MAP]
       │                           │                        │
       │ listObjects(inputPath)    │                        │
       │ → ["key1.txt","key2.txt",...]                      │
       │                           │                        │
       │ split into 4 chunks:      │                        │
       │   chunk0: [key1.txt]      │                        │
       │   chunk1: [key2.txt]      │                        │
       │   chunk2: [key3.txt]      │                        │
       │   chunk3: [key4.txt]      │                        │
       │                           │                        │
       │ INSERT tasks (MAP, IDLE)  │                        │
       │ × 4                       │                        │
       │                           │                        │
       │ UPDATE job → MAP_PHASE    │                        │
       │                           │                        │
       │ for each task:            │                        │
       │   create batch/v1/Job     │                        │
       │   name=worker-map-{taskId}│                        │
       │   env: TASK_TYPE=MAP      │                        │
       │   env: INPUT_SPLIT=key.txt│                        │
       │   env: CODE_PATH=users/.. │                        │
       ├──────────────────────────►│                        │
       │                           │ schedules pod          │
       │                           │ pulls worker image     │
       │                           ├───────────────────────►│
       │                           │                        │
       │                           │                        │ reportInProgress()
       │◄──────────────────────────────────────────────────┤
       │ UPDATE task → IN_PROGRESS │                        │
       │                           │                        │
       │                           │                        │ startHeartbeat()
       │                           │                        │ downloadToFile(codePath)
       │                           │                        │ URLClassLoader(code.jar)
       │                           │                        │ ServiceLoader.load(Mapper)
       │                           │                        │
       │                           │                        │ for each key in INPUT_SPLIT:
       │                           │                        │   downloadAsString(key)
       │                           │                        │   for each line:
       │                           │                        │     mapper.map(key:i, line, emit)
       │                           │                        │
       │                           │                        │ write intermediate to:
       │                           │                        │ temp/{jobId}/map-{taskId}/part-0.txt
       │                           │                        │
       │                           │                        │ reportCompleted(outputKey)
       │◄──────────────────────────────────────────────────┤
       │ UPDATE task → COMPLETED   │                        │
       │ task.outputLocation = ... │                   [pod exits 0]
       │                           │                        │
       │ countCompleted == 4? → startReducePhase()

13.3 Reduce Phase — Step by Step

[Manager]                    [Kubernetes]            [Worker Pod - REDUCE]
    │                              │                          │
    │ collect outputLocations      │                          │
    │ from all 4 MAP tasks:        │                          │
    │   [temp/job/map-0/part-0.txt,│                          │
    │    temp/job/map-1/part-0.txt,│                          │
    │    temp/job/map-2/part-0.txt,│                          │
    │    temp/job/map-3/part-0.txt]│                          │
    │                              │                          │
    │ split across 2 reducers:     │                          │
    │   reducer0: [map-0, map-2]   │                          │
    │   reducer1: [map-1, map-3]   │                          │
    │                              │                          │
    │ INSERT tasks (REDUCE, IDLE)  │                          │
    │ × 2                          │                          │
    │                              │                          │
    │ UPDATE job → REDUCE_PHASE    │                          │
    │                              │                          │
    │ CREATE batch/v1/Job ×2       │                          │
    │   env: TASK_TYPE=REDUCE      │                          │
    │   env: INPUT_SPLIT=temp/..., │                          │
    │        temp/...              │                          │
    ├─────────────────────────────►│                          │
    │                              ├─────────────────────────►│
    │                              │                          │
    │                              │                          │ downloadAsString(map-0)
    │                              │                          │ downloadAsString(map-2)
    │                              │                          │
    │                              │                          │ parse lines → group by key
    │                              │                          │ TreeMap {
    │                              │                          │   "the" → ["1","1","1","1",...],
    │                              │                          │   "and" → ["1","1","1",...],
    │                              │                          │   ...
    │                              │                          │ }
    │                              │                          │
    │                              │                          │ ServiceLoader.load(Reducer)
    │                              │                          │
    │                              │                          │ for each key:
    │                              │                          │   reducer.reduce(key, values, emit)
    │                              │                          │   // WordCountReducer sums them: 892
    │                              │                          │
    │                              │                          │ write to:
    │                              │                          │ users/{id}/results/{jobId}/reduce-0.txt
    │                              │                          │
    │                              │                          │ reportCompleted(outputKey)
    │◄─────────────────────────────────────────────────────── │
    │ UPDATE task → COMPLETED      │                     [pod exits 0]
    │                              │                          │
    │ countCompleted == 2 → UPDATE job → COMPLETED
    │ generate presigned URLs for output files

13.4 Polling for Results

CLI polls every 5 seconds while watching:

GET /api/v1/jobs/{jobId}
  → UI → Manager → getJobStatus()
  → if COMPLETED: listObjects(outputPath) → presignedGetUrl() ×N
  → returns { status: COMPLETED, outputUrls: ["http://minio:9000/...?X-Amz-Signature=..."] }

mr results {jobId}:
  → downloads each presigned URL directly (not through the API, straight from MinIO)
  → concatenates all output files
  → parses as "word\tcount" pairs
  → sorts by count descending
  → renders table:

  Results for job abc-123-...
  156 unique entries · 12,450 total

  KEY           COUNT   FREQUENCY
  ──────────────────────────────────────────────────────
  the             892   ██████████████████████████████░
  and             654   █████████████████████░░░░░░░░░
  a               421   █████████████░░░░░░░░░░░░░░░░░
  to              398   ████████████░░░░░░░░░░░░░░░░░░
  ...

---

14. Failure Handling & Retries

There are three layers of retry in this system:

Layer 1: Kubernetes backoffLimit

spec:
  backoffLimit: 3   # K8s will retry the pod up to 3 times if it exits non-zero

If the pod crashes without calling reportFailed() (e.g., OOMKilled, network error before any code runs), Kubernetes automatically restarts it. The restartPolicy: Never means don't restart the same pod — Kubernetes creates a new pod.

Layer 2: Application-level retry in Manager (onTaskFailed)

When a worker reports FAILED back to the manager:

retry_count = 0: delete old K8s job → launch new one → task stays IN_PROGRESS
retry_count = 1: same
retry_count = 2: same
retry_count = 3: give up → mark job as FAILED, set errorMessage

Layer 3: Heartbeat watchdog

Every 15 seconds, HeartbeatWatchdog queries:
  SELECT * FROM tasks
  WHERE status = 'IN_PROGRESS'
  AND last_heartbeat < NOW() - INTERVAL '30 seconds'

For each stale task:
  → Treats it as FAILED → triggers Layer 2 retry logic

The heartbeat timeout (30s) vs watchdog interval (15s) means a dead worker is detected within 15–45 seconds of its last heartbeat.

What happens when a reduce worker fails after the map phase completed?

• The reduce task is retried (up to 3 times)
• The intermediate map outputs are still in MinIO (temp/{jobId}/map-{taskId}/)
• The new reduce worker downloads the same intermediate files and reruns the reduce
• The map phase is NOT rerun

What happens when a map worker fails?

• The map task is retried
• The input data is still in MinIO (users/{userId}/raw/...)
• The new map worker downloads the same input and reruns map from scratch

---

15. Dockerfiles

All three services use the same multi-stage build pattern. The idea: compile in a fat JDK image, run in a slim JRE image.

Stage 1: Build

FROM eclipse-temurin:21-jdk-alpine AS builder
WORKDIR /build
# copy root pom + all module dirs
COPY pom.xml .
COPY common/ common/
COPY manager-service/ manager-service/
# build only what we need (common is a dependency of manager-service)
RUN mvn -pl common,manager-service -am package -DskipTests -q

The -pl common,manager-service -am flag tells Maven to build only these modules (plus their dependencies with -am). Without -am, it would fail because it can't find common's compiled classes.

Stage 2: Runtime

FROM eclipse-temurin:21-jre-alpine
WORKDIR /app

# non-root user for security — if someone gets RCE they're "appuser" not "root"
RUN addgroup -S appgroup && adduser -S appuser -G appgroup
USER appuser

COPY --from=builder /build/manager-service/target/*.jar app.jar
EXPOSE 8081
ENTRYPOINT ["java", "-jar", "app.jar"]

Why Alpine? The Alpine Linux base is ~5MB vs ~30MB for Debian-based images. eclipse-temurin:21-jre-alpine is roughly 90MB vs 300MB+ for a full JDK image. Smaller images = faster pod starts and less storage.

Per-service differences:

Service	-pl flag	EXPOSE	ENTRYPOINT
ui-service	common,ui-service	8080	java -jar app.jar
manager-service	common,manager-service	8081	java -jar app.jar
worker	common,worker	(none)	java -jar app.jar

Worker has no EXPOSE because it never receives incoming connections — it only makes outbound HTTP calls to the manager and MinIO.

---

16. Kubernetes Setup

All manifests live in k8s/. There's a deploy.sh that applies them in the right order (namespace first, then infra, then apps).

16.1 Namespace

# k8s/namespace/namespace.yaml
apiVersion: v1
kind: Namespace
metadata:
  name: mapreduce

Everything goes in this namespace. Services can refer to each other by their service name (e.g. manager-service, postgres, minio, keycloak) — Kubernetes DNS resolves these within the namespace.

16.2 Infrastructure

PostgreSQL

# k8s/postgres/postgres-statefulset.yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: postgres
  namespace: mapreduce
spec:
  replicas: 1
  serviceName: postgres
  selector:
    matchLabels:
      app: postgres
  template:
    spec:
      containers:
        - name: postgres
          image: postgres:16-alpine
          env:
            - name: POSTGRES_DB       → from ConfigMap
            - name: POSTGRES_USER     → from Secret
            - name: POSTGRES_PASSWORD → from Secret
          ports:
            - containerPort: 5432
          volumeMounts:
            - mountPath: /var/lib/postgresql/data
              name: postgres-data
  volumeClaimTemplates:
    - metadata: { name: postgres-data }
      spec:
        accessModes: [ReadWriteOnce]
        resources:
          requests:
            storage: 5Gi

StatefulSet for Postgres gives it a stable pod name (postgres-0) and a persistent volume that survives pod restarts. If the pod dies and is rescheduled, it reconnects to the same volume.

The Service for Postgres is a ClusterIP on port 5432 — only reachable from inside the cluster (manager service).

MinIO

# k8s/minio/minio-statefulset.yaml
# Similar pattern to Postgres
image: quay.io/minio/minio:latest
command: ["minio", "server", "/data", "--console-address", ":9001"]
ports:
  - containerPort: 9000   # S3 API
  - containerPort: 9001   # Console

# Service: NodePort
ports:
  - name: api     port: 9000  nodePort: 30900
  - name: console port: 9001  nodePort: 30901

MinIO is exposed via NodePort so you can access the web console from your browser at http://{minikube-ip}:30901.

Keycloak

# k8s/keycloak/keycloak-deployment.yaml
image: quay.io/keycloak/keycloak:25.0
args: ["start-dev"]   # development mode, no TLS
env:
  - KC_DB: postgres
  - KC_DB_URL: jdbc:postgresql://postgres:5432/keycloak
  - KC_DB_USERNAME / KC_DB_PASSWORD → from Secret
  - KEYCLOAK_ADMIN / KEYCLOAK_ADMIN_PASSWORD → from Secret

# Service: NodePort :30180

Keycloak stores its own data in the same PostgreSQL instance but a different database (keycloak vs mapreduce).

16.3 Application Services

Manager Service

# k8s/manager/manager-statefulset.yaml
kind: StatefulSet
replicas: 2
serviceName: manager-service
spec:
  containers:
    - name: manager-service
      image: mapreduce/manager-service:latest
      ports:
        - containerPort: 8081
      envFrom:
        - configMapRef: { name: manager-config }
        - secretRef:    { name: manager-secret }
      readinessProbe:
        httpGet: { path: /actuator/health/readiness, port: 8081 }
        initialDelaySeconds: 60
        periodSeconds: 10
      livenessProbe:
        httpGet: { path: /actuator/health/liveness, port: 8081 }
        initialDelaySeconds: 90
        periodSeconds: 20
        failureThreshold: 5
      resources:
        requests: { cpu: 250m, memory: 512Mi }
        limits:   { cpu: 1,    memory: 1Gi  }
  serviceAccountName: manager-sa

The serviceAccountName: manager-sa is what grants the manager permission to create K8s Jobs. Without it, the Fabric8 client would get a 403 when trying to create worker pods.

The Service for manager is a headless ClusterIP (no NodePort) — only the UI service can reach it, and only from inside the cluster.

Manager RBAC

# k8s/manager/manager-rbac.yaml

apiVersion: v1
kind: ServiceAccount
metadata:
  name: manager-sa
  namespace: mapreduce

---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: manager-job-role
  namespace: mapreduce
rules:
  - apiGroups: ["batch"]
    resources: ["jobs"]
    verbs: ["create", "get", "list", "watch", "delete"]
  - apiGroups: [""]
    resources: ["pods"]
    verbs: ["get", "list", "watch"]

---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: manager-job-rolebinding
  namespace: mapreduce
subjects:
  - kind: ServiceAccount
    name: manager-sa
    namespace: mapreduce
roleRef:
  kind: Role
  name: manager-job-role
  apiGroup: rbac.authorization.k8s.io

This is a namespaced Role (not ClusterRole) — the manager can only create jobs in the mapreduce namespace, not cluster-wide. Least-privilege principle.

UI Service

# k8s/ui/ui-deployment.yaml
kind: Deployment
replicas: 1
spec:
  containers:
    - name: ui-service
      image: mapreduce/ui-service:latest
      ports:
        - containerPort: 8080
      resources:
        requests: { cpu: 125m, memory: 256Mi }
        limits:   { cpu: 500m, memory: 512Mi }
      readinessProbe:
        httpGet: { path: /actuator/health/readiness, port: 8080 }
        initialDelaySeconds: 60
      livenessProbe:
        httpGet: { path: /actuator/health/liveness, port: 8080 }

# Service: NodePort :30080

UI uses Deployment (not StatefulSet) because it's stateless — any replica is identical. It only has 1 replica here but you could scale to 2+ easily.

16.4 ConfigMaps and Secrets

Each service has its config split between a ConfigMap (non-sensitive) and a Secret (sensitive). Example for Manager:

# manager-configmap.yaml
data:
  DB_HOST:              postgres
  DB_PORT:              "5432"
  DB_NAME:              mapreduce
  MINIO_ENDPOINT:       http://minio:9000
  MINIO_PUBLIC_ENDPOINT: http://192.168.49.2:30900
  MINIO_BUCKET:         mapreduce
  K8S_NAMESPACE:        mapreduce
  WORKER_IMAGE:         mapreduce/worker:latest
  MANAGER_INTERNAL_URL: http://manager-service:8081
  KEYCLOAK_ISSUER_URI:  http://keycloak:8080/realms/mapreduce
  KEYCLOAK_JWK_SET_URI: http://keycloak:8080/realms/mapreduce/protocol/openid-connect/certs
  DEFAULT_MAP_TASKS:    "4"

# manager-secret.yaml (values are base64-encoded)
data:
  DB_USER:          bWFuYWdlcg==      # "manager"
  DB_PASSWORD:      ...
  MINIO_ACCESS_KEY: ...
  MINIO_SECRET_KEY: ...

---

17. Docker Compose (local dev)

# docker-compose.yml
services:

  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB:       mapreduce
      POSTGRES_USER:     mapreduce
      POSTGRES_PASSWORD: mapreduce-secret
    ports: ["5432:5432"]
    volumes: [postgres-data:/var/lib/postgresql/data]

  minio:
    image: quay.io/minio/minio:latest
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER:     minioadmin
      MINIO_ROOT_PASSWORD: minioadmin-secret
    ports: ["9000:9000", "9001:9001"]
    volumes: [minio-data:/data]

  keycloak:
    image: quay.io/keycloak/keycloak:25.0
    command: start-dev
    environment:
      KC_DB:              postgres
      KC_DB_URL:          jdbc:postgresql://postgres:5432/keycloak
      KC_DB_USERNAME:     mapreduce
      KC_DB_PASSWORD:     mapreduce-secret
      KEYCLOAK_ADMIN:     admin
      KEYCLOAK_ADMIN_PASSWORD: admin-secret
    ports: ["8180:8080"]
    depends_on: [postgres]

  manager:
    build: { context: ., dockerfile: manager-service/Dockerfile }
    environment:
      DB_HOST:              postgres
      DB_USER:              mapreduce
      DB_PASSWORD:          mapreduce-secret
      MINIO_ENDPOINT:       http://minio:9000
      MINIO_PUBLIC_ENDPOINT: http://localhost:9000
      MINIO_ACCESS_KEY:     minioadmin
      MINIO_SECRET_KEY:     minioadmin-secret
      KEYCLOAK_ISSUER_URI:  http://keycloak:8080/realms/mapreduce
      K8S_NAMESPACE:        mapreduce
      WORKER_IMAGE:         mapreduce/worker:latest
      MANAGER_INTERNAL_URL: http://manager:8081
    ports: ["8081:8081"]
    depends_on: [postgres, minio, keycloak]

  ui:
    build: { context: ., dockerfile: ui-service/Dockerfile }
    environment:
      MANAGER_BASE_URL:      http://manager:8081
      KEYCLOAK_ISSUER_URI:   http://keycloak:8080/realms/mapreduce
      KEYCLOAK_ADMIN_BASE_URL: http://keycloak:8080
      KEYCLOAK_ADMIN_USERNAME: admin
      KEYCLOAK_ADMIN_PASSWORD: admin-secret
      KEYCLOAK_REALM:        mapreduce
    ports: ["8080:8080"]
    depends_on: [manager]

volumes:
  postgres-data:
  minio-data:

With Docker Compose, workers can't actually be launched in Kubernetes (there's no K8s running). To test locally you'd either mock the K8s launcher or run a local Kubernetes (Minikube, Kind, etc.) and point the manager at it.

---

18. Database Schema

Flyway runs SQL migrations in order when the manager starts. The files live at:

manager-service/src/main/resources/db/migration/
  V1__create_jobs_and_tasks.sql
  V2__create_file_metadata.sql

V1 — jobs and tasks tables

-- jobs table
CREATE TABLE jobs (
    job_id           UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id          VARCHAR(255) NOT NULL,
    status           VARCHAR(50)  NOT NULL,       -- stores enum name: MAP_PHASE, etc.
    code_path        VARCHAR(500),
    input_path       VARCHAR(500),
    output_path      VARCHAR(500),
    num_map_tasks    INT          NOT NULL DEFAULT 4,
    num_reduce_tasks INT          NOT NULL DEFAULT 2,
    error_message    TEXT,
    created_at       TIMESTAMP    NOT NULL DEFAULT NOW(),
    updated_at       TIMESTAMP    NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_jobs_user_id ON jobs(user_id);
CREATE INDEX idx_jobs_status  ON jobs(status);

-- Auto-update updated_at on any row change
CREATE OR REPLACE FUNCTION update_updated_at_column()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = NOW();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_jobs_updated_at
    BEFORE UPDATE ON jobs
    FOR EACH ROW
    EXECUTE FUNCTION update_updated_at_column();

-- tasks table
CREATE TABLE tasks (
    task_id         UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
    job_id          UUID         NOT NULL REFERENCES jobs(job_id) ON DELETE CASCADE,
    task_type       VARCHAR(10)  NOT NULL,   -- MAP or REDUCE
    status          VARCHAR(20)  NOT NULL,   -- IDLE, IN_PROGRESS, COMPLETED, FAILED
    worker_pod_id   VARCHAR(255),            -- K8s Job name: worker-map-{taskId}
    input_split     TEXT,                    -- comma-separated MinIO keys
    output_location TEXT,                    -- MinIO path written by worker
    retry_count     INT          NOT NULL DEFAULT 0,
    last_heartbeat  TIMESTAMP,
    error_message   TEXT,
    created_at      TIMESTAMP    NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMP    NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_tasks_job_id ON tasks(job_id);
CREATE INDEX idx_tasks_status  ON tasks(status);

CREATE TRIGGER trg_tasks_updated_at
    BEFORE UPDATE ON tasks
    FOR EACH ROW
    EXECUTE FUNCTION update_updated_at_column();

V2 — file_metadata table

CREATE TABLE file_metadata (
    file_id       UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id       VARCHAR(255) NOT NULL,
    file_type     VARCHAR(10)  NOT NULL,   -- DATA or CODE
    original_name VARCHAR(500),
    storage_path  VARCHAR(500) NOT NULL,   -- MinIO object key
    size_bytes    BIGINT,
    created_at    TIMESTAMP    NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_file_metadata_user_id   ON file_metadata(user_id);
CREATE INDEX idx_file_metadata_file_type ON file_metadata(file_type);

Why store file metadata in Postgres at all? When a user submits a job with dataId and codeId, the manager needs to verify:

1. Those files actually exist
2. They belong to this user (not someone else's files)
3. What their MinIO paths are (to give to the workers)

Without the metadata table, you'd have to list MinIO objects to find files, which is slow and doesn't give you ownership info.

---

19. Monitoring & Health

Spring Actuator

Both services expose Actuator endpoints:

/actuator/health             → combined health status
/actuator/health/readiness   → is the app ready to handle requests?
/actuator/health/liveness    → is the app alive (not deadlocked)?
/actuator/prometheus         → Prometheus metrics scrape endpoint

Readiness vs Liveness:

• Readiness: "Can I take traffic right now?" — Spring waits until Flyway migrations finish, Keycloak JWKS is loaded, etc. K8s won't route traffic until this returns 200.
• Liveness: "Am I still functioning?" — if this starts returning errors, K8s kills the pod and restarts it.

The initial delays (initialDelaySeconds: 60) give the JVM and Spring time to start up before K8s starts checking.

Micrometer + Prometheus

Micrometer is Spring Boot's metrics facade. It auto-instruments:

• JVM: heap usage, GC pauses, thread counts
• HTTP: request count, error rate, response time (by endpoint)
• HikariCP: connection pool usage (for manager only)
• Spring Data JPA: query execution times

To actually see these metrics you'd run Prometheus (scraping /actuator/prometheus every 15s) and Grafana (displaying dashboards). Those aren't in this repo but you could add them to Docker Compose or Kubernetes.

Swagger UI

The UI service has Springdoc OpenAPI configured:

@Configuration
@OpenAPIDefinition(info = @Info(title = "MapReduce API", version = "1.0"))
@SecurityScheme(
    name = "Bearer Auth",
    type = SecuritySchemeType.HTTP,
    scheme = "bearer",
    bearerFormat = "JWT"
)
public class OpenApiConfig { }

This auto-generates OpenAPI 3 spec from the controller annotations and exposes:

• http://host:30080/swagger-ui.html — interactive UI
• http://host:30080/v3/api-docs — raw JSON spec

You can use the Swagger UI to test API calls directly in the browser (with Bearer token pasted in).

---

20. Complete End-to-End Flow

Here's the entire numbered flow for mr run data.txt wordcount.jar --reducers 2 --watch 3:

[Upload Phase]
 [1]  CLI: POST /api/v1/data multipart (data.txt)
 [2]  UI:  validates JWT, extracts userId → POST /internal/files/data
 [3]  Manager: generates key = users/{userId}/raw/{uuid}_data.txt
 [4]  Manager: MinIO.putObject(key, stream, size)
 [5]  Manager: INSERT file_metadata → returns {id: dataUUID}
 [6]  CLI: saves session.lastDataId = dataUUID
 [7]  CLI: POST /api/v1/code multipart (wordcount.jar) → same flow → codeUUID

[Job Submission]
 [8]  CLI: POST /api/v1/jobs {dataId, codeId, numReducers: 2}
 [9]  UI:  proxy to Manager POST /internal/jobs
 [10] Manager: lookup file_metadata for dataId → get inputPath
 [11] Manager: INSERT jobs (INITIALIZING) → jobId
 [12] CLI: receives {jobId: "abc-123"} → start polling

[Map Phase - async]
 [13] Manager: listObjects(inputPath) → ["key1.txt"]
 [14] Manager: split into 4 chunks (each gets ~1/4 of keys)
 [15] Manager: INSERT 4 tasks (MAP, IDLE, inputSplit=...)
 [16] Manager: UPDATE job → MAP_PHASE
 [17] Manager: k8sLauncher.launchWorker() × 4
      K8s creates 4 batch/v1/Jobs → 4 pods start

[Worker Execution - per pod]
 [18] Worker: reportInProgress() → manager updates task IN_PROGRESS
 [19] Worker: starts heartbeat thread (every 10s)
 [20] Worker: downloads code.jar from MinIO
 [21] Worker: URLClassLoader(code.jar) → ServiceLoader.load(Mapper)
              finds WordCountMapper via META-INF/services/...Mapper
 [22] Worker: downloads input file(s) from MinIO
 [23] Worker: for each line → mapper.map(key, line, emit)
              WordCountMapper: splits by whitespace, emits ("word", "1") per word
 [24] Worker: sorts intermediate k-v in TreeMap
 [25] Worker: writes "word\t1\nword\t1\n..." to MinIO
              key: temp/abc-123/map-{taskId}/part-0.txt
 [26] Worker: reportCompleted(outputKey)
              → manager updates task COMPLETED, stores outputLocation
 [27] All 4 workers complete → manager sees countCompleted == 4

[Reduce Phase - async]
 [28] Manager: collects 4 output paths (temp/.../part-0.txt × 4)
 [29] Manager: distributes across 2 reducers:
              reducer0: [map-0/part-0.txt, map-2/part-0.txt]
              reducer1: [map-1/part-0.txt, map-3/part-0.txt]
 [30] Manager: INSERT 2 tasks (REDUCE, IDLE, inputSplit=...)
 [31] Manager: UPDATE job → REDUCE_PHASE
 [32] Manager: k8sLauncher.launchWorker() × 2

[Reduce Worker Execution]
 [33] Worker: downloads code.jar (same jar as before)
 [34] Worker: ServiceLoader.load(Reducer) → finds WordCountReducer
 [35] Worker: downloads 2 intermediate files
 [36] Worker: parses "word\t1" lines → groups by key:
              { "the" → ["1","1","1",...], "and" → ["1","1",...], ... }
 [37] Worker: for each key → reducer.reduce(key, values, emit)
              WordCountReducer: sum all "1"s → emit ("the", "892")
 [38] Worker: writes "the\t892\nand\t654\n..." to MinIO
              key: users/{userId}/results/abc-123/reduce-0.txt
 [39] Worker: reportCompleted(outputKey)
 [40] Both reduce workers complete → manager sees countCompleted == 2
 [41] Manager: UPDATE job → COMPLETED

[Results]
 [42] CLI poll: GET /api/v1/jobs/abc-123
 [43] Manager: listObjects("users/{userId}/results/abc-123/")
              → ["reduce-0.txt", "reduce-1.txt"]
 [44] Manager: presignedGetUrl("reduce-0.txt", 3600) × 2
 [45] Returns {status: COMPLETED, outputUrls: ["http://...", "http://..."]}
 [46] CLI prints: "Job completed. Results: mr results abc-123"

[mr results abc-123]
 [47] CLI: GET each presigned URL directly from MinIO
 [48] CLI: concatenates content of reduce-0.txt + reduce-1.txt
 [49] CLI: parses "word\tcount" lines
 [50] CLI: all values are numbers → sort by count descending
 [51] CLI: renders table with bar chart

  Results for job abc-123-...
  156 unique entries · 12,450 total

  KEY           COUNT   FREQUENCY
  ─────────────────────────────────────────────────────
  the             892   ██████████████████████████████
  and             654   █████████████████████░░░░░░░░░
  a               421   █████████████░░░░░░░░░░░░░░░░░
  ...

---

Summary Table

	ui-service	manager-service	worker	cli
Framework	Spring Boot 3.3.4	Spring Boot 3.3.4	Spring Boot 3.3.4 (CommandLineRunner)	Plain Java 21
Auth	OAuth2 Resource Server	OAuth2 Resource Server (split chains)	None (batch pod)	JWT Bearer header
Database	—	PostgreSQL + JPA + Flyway	—	~/.mr/session.json
Storage	—	MinIO SDK (2 clients)	MinIO SDK	—
K8s client	—	Fabric8 v6.13.1	—	—
CLI framework	—	—	—	picocli v4.7.6
JSON	Jackson (via Spring)	Jackson (via Spring)	Jackson (via Spring)	Jackson v2.17.2
DTO mapping	MapStruct	MapStruct	—	—
Metrics	Micrometer + Prometheus	Micrometer + Prometheus	—	—
API docs	Springdoc OpenAPI v2.6	—	—	picocli --help
Port	8080 (NodePort 30080)	8081 (ClusterIP only)	no server port	local only
K8s kind	Deployment (1 replica)	StatefulSet (2 replicas)	batch/v1/Job (dynamic)	local binary
Docker base	eclipse-temurin:21-jre-alpine	eclipse-temurin:21-jre-alpine	eclipse-temurin:21-jre-alpine	fat JAR
Lifecycle	long-running server	long-running server	starts → runs → exits	interactive shell / one-shot