TIL

Remote Chunking

Remote Chunking 개요

원격 파티셔닝과의 차이

• 원격 파티셔닝: Manager가 작업 지침(StepExecutionRequest — ID만 포함)을 Worker에게 보내고, Worker가 직접 데이터를 조회하여 처리
• 원격 청킹: Manager가 직접 데이터를 읽어 청크 단위로 묶은 뒤, 데이터 자체를 Worker에게 전송 — Worker는 전달받은 데이터를 처리(Process/Write)하고 결과를 회신
• 역할 분담:
  • Manager = **ChunkProvider** (데이터 읽기 담당)
  • Worker = **ChunkProcessor** (데이터 처리·쓰기 담당)
• 핵심 메시지: Manager → Worker로 ChunkRequest(처리할 데이터 포함), Worker → Manager로 ChunkResponse(처리 결과 요약)

사용 시나리오

• Processing/Writing 부하가 극심할 때: 데이터를 읽는 것은 빠르지만 처리(Processor)나 쓰기(Writer)가 복잡하고 느린 경우
• 데이터 소스 분할이 불가능할 때:
  • 단일 스트림 데이터
    • ex) 실시간 로그 피드
  • 순서가 중요한 메시지 큐
    • ex) 금융 거래 이벤트 (입금 -> 출금)
    • 여러 Worker가 각자 큐를 읽으면 순서가 깨질 수 있기에 Manager가 순서대로 읽어서 청크로 넘겨야 한다.
  • 파티셔닝 키가 없는 레거시 데이터 소스 등
    • ex) csv 파일이나 고정 플랫 파일에서 ID 같은 식별자가 없을 때

한계점

• 네트워크 대역폭 소모: 실제 데이터를 전송하므로 데이터 건수가 많거나 크기가 크면 통신 자체가 병목이 될 수 있다
• Manager 읽기 병목: Manager 혼자 모든 데이터를 읽어야 하므로, 읽기 자체가 느리면 원격 청킹의 효과가 없다
• 복잡성: 원격 파티셔닝과 마찬가지로 Message Broker와 Spring Integration 설정이 필수적이다

Manager Node 구성

• @EnableBatchIntegration 선언 시 자동 주입되는 RemoteChunkingManagerStepBuilderFactory를 사용한다
• 원격 파티셔닝과의 구조적 차이: Manager Step에 reader()가 있다 — Manager가 직접 데이터를 읽는다

@Bean
public Step managerStep() {
    return remoteChunkingManagerStepBuilderFactory
            .get("managerStep")
            .chunk(3)
            .reader(dataReader())                        // Manager가 직접 데이터를 읽는다
            .outputChannel(outboundChunksToWorkers())    // Worker에게 ChunkRequest를 보낼 채널
            .inputChannel(inboundRepliesFromWorkers())   // Worker로부터 ChunkResponse를 받을 채널
            .build();
}

내부 메커니즘: ChunkMessageChannelItemWriter

• RemoteChunkingManagerStepBuilder는 내부적으로 ItemWriter로 ChunkMessageChannelItemWriter를 사용한다
• 이 Writer는 DB나 파일에 쓰는 것이 아니라, reader가 읽은 데이터 청크를 ChunkRequest 메시지로 포장하여 outputChannel로 전송한다
• 또한 inputChannel을 통해 모든 Worker로부터 응답(ChunkResponse)이 도착할 때까지 대기(getNextResult())하여 작업 완료를 보장한다

// ChunkRequest — 데이터 자체가 네트워크를 통해 이동한다
public class ChunkRequest<T> implements Serializable {
    private final long jobId;                       // JobInstance ID
    private final Chunk<? extends T> items;         // 실제 데이터 아이템들이 담긴 청크
    private final StepContribution stepContribution;
    private final int sequence;                     // 청크 순번
}

채널과 IntegrationFlow

• 요청은 chunk-request 토픽, 응답은 chunk-response 토픽을 사용한다

@Bean
public DirectChannel outboundChunksToWorkers() { return new DirectChannel(); }

@Bean // 비동기 수신을 위해 QueueChannel 사용
public QueueChannel inboundRepliesFromWorkers() { return new QueueChannel(); }

@Bean // outboundChunksToWorkers 채널 → Kafka chunk-request 토픽
public IntegrationFlow outboundChunkFlow(KafkaTemplate<Long, ChunkRequest> kafkaTemplate) {
    KafkaProducerMessageHandler<Long, ChunkRequest> handler =
            new KafkaProducerMessageHandler<>(kafkaTemplate);
    handler.setTopicExpression(new LiteralExpression("chunk-request"));
    // ChunkRequest의 sequence 번호 기반으로 Kafka 파티션(Worker)에 분배
    handler.setPartitionIdExpression(new FunctionExpression<>(message -> {
        ChunkRequest req = (ChunkRequest) message.getPayload();
        return req.getSequence() % partitionSize;
    }));
    return IntegrationFlow.from(outboundChunksToWorkers()).handle(handler).get();
}

@Bean // Kafka chunk-response 토픽 → inboundRepliesFromWorkers 채널
public IntegrationFlow inboundReplyFlow(ConsumerFactory<Long, ChunkResponse> cf) {
    return IntegrationFlow
            .from(Kafka.messageDrivenChannelAdapter(cf, "chunk-response"))
            .channel(inboundRepliesFromWorkers())
            .get();
}

Worker Node 구성

• @EnableBatchIntegration 선언 시 자동 주입되는 RemoteChunkingWorkerBuilder를 사용한다
• 원격 파티셔닝과의 핵심 차이: Worker는 Step이 아니라 IntegrationFlow를 생성한다 — Worker는 스텝을 실행하는 것이 아니라 메시지 처리 흐름을 타기 때문이다
• itemProcessor()와 itemWriter()를 설정하는 것이 핵심 — 원격 청킹의 목적 자체가 이 두 단계의 부하를 Worker에 분산시키는 것이다

@Bean
public IntegrationFlow workerIntegrationFlow() {
    return remoteChunkingWorkerBuilder
            .inputChannel(inboundChunkRequestsFromManager())   // ChunkRequest 수신 채널
            .outputChannel(outboundRepliesToManager())          // ChunkResponse 전송 채널
            .itemProcessor(processor())                        // 수신된 청크의 각 아이템 처리
            .itemWriter(writer())                              // 처리 결과 기록
            .build();
}

내부 메커니즘: ChunkProcessorChunkHandler

• RemoteChunkingWorkerBuilder는 내부적으로 itemProcessor와 itemWriter를 사용하는 SimpleChunkProcessor를 만들고, 이를 ChunkProcessorChunkHandler에 설정한다
• ChunkRequest 메시지가 inputChannel로 들어오면, IntegrationFlow가 ChunkProcessorChunkHandler.handleChunk()로 전달한다
• handleChunk()는 SimpleChunkProcessor를 사용하여 청크(ChunkRequest.items)를 처리(process/write)하고, 결과를 담은 ChunkResponse를 생성하여 outputChannel로 내보낸다

// ChunkResponse — Worker의 청크 처리 결과
public class ChunkResponse implements Serializable {
    private final StepContribution stepContribution; // 핵심: 처리 통계(읽은 수, 쓴 수, 스킵 수 등)
    private final Long jobId;
    private final boolean status;    // 성공/실패 여부
    private final String message;    // 오류 메시지 (선택적)
    private final int sequence;      // 처리한 청크의 순번
}

• stepContribution - Worker가 처리한 단일 청크의 모든 처리 기록(읽은 수, 쓴 수, 필터링/스킵 수, 상태 등)이 담겨 있다
• Manager는 ChunkResponse를 수신할 때마다 각 stepContribution의 값을 자신의 메인 StepExecution 상태에 합산(aggregate)하여, 최종적으로 Job의 전체 처리 통계와 완료 상태를 결정한다

채널과 IntegrationFlow

@Bean
public QueueChannel inboundChunkRequestsFromManager() { return new QueueChannel(); }

@Bean
public DirectChannel outboundRepliesToManager() { return new DirectChannel(); }

@Bean // Kafka chunk-request 토픽 → inboundChunkRequestsFromManager 채널
public IntegrationFlow inboundChunkRequestFlow(ConsumerFactory<Long, ChunkRequest> cf) {
    return IntegrationFlow
            .from(Kafka.messageDrivenChannelAdapter(cf, "chunk-request"))
            .channel(inboundChunkRequestsFromManager())
            .get();
}

@Bean // outboundRepliesToManager 채널 → Kafka chunk-response 토픽
public IntegrationFlow outboundResponseFlow(KafkaTemplate<Long, ChunkResponse> kafkaTemplate) {
    KafkaProducerMessageHandler<Long, ChunkResponse> handler =
            new KafkaProducerMessageHandler<>(kafkaTemplate);
    handler.setTopicExpression(new LiteralExpression("chunk-response"));
    return IntegrationFlow.from(outboundRepliesToManager()).handle(handler).get();
}

직렬화 및 환경 설정

• ChunkRequest(내부에 데이터 Item 포함)와 ChunkResponse를 Kafka로 전송하려면 전용 Serializer/Deserializer가 필요하다
• 원격 파티셔닝의 StepExecution 직렬화와 동일한 원리 — Spring의 DefaultSerializer/DefaultDeserializer를 활용한 커스텀 SerDes 구현
• application.yml에서 각 프로파일의 producer/consumer에 커스텀 SerDes 클래스를 지정한다
  • Manager: ChunkRequest 전송(producer → ChunkRequestSerializer), ChunkResponse 수신(consumer → ChunkResponseDeserializer)
  • Worker: ChunkRequest 수신(consumer → ChunkRequestDeserializer), ChunkResponse 전송(producer → ChunkResponseSerializer)

# Manager 프로파일
spring:
  config:
    activate:
      on-profile: remote-chunking-manager
  kafka:
    producer:
      value-serializer: com.example.ChunkRequestSerDes$ChunkRequestSerializer
    consumer:
      value-deserializer: com.example.ChunkResponseSerDes$ChunkResponseDeserializer
      group-id: manager
---
# Worker 프로파일
spring:
  config:
    activate:
      on-profile: remote-chunking-worker
  kafka:
    producer:
      value-serializer: com.example.ChunkResponseSerDes$ChunkResponseSerializer
    consumer:
      value-deserializer: com.example.ChunkRequestSerDes$ChunkRequestDeserializer
      group-id: worker