基于AKKA实现Kafka Source经KPL发送至Kinesis Sink的问题

1. 配置KPLFlowSettings #

先初始化KPLFlowSettings，指定目标Kinesis流名称、AWS区域，以及KPL的核心批量、重试等配置。AWS认证会自动读取环境变量、本地凭证文件或IAM角色权限（若在AWS托管环境运行）。

import com.github.j5ik2o.akka.kinesis.kpl.KPLFlowSettings;
import software.amazon.awssdk.regions.Region;

KPLFlowSettings kplFlowSettings = KPLFlowSettings.builder()
    .streamName("你的目标Kinesis流名称")
    .region(Region.US_EAST_1) // 替换为实际使用的AWS区域
    .maxBatchSize(500) // 可选：单次批量发送的最大记录数
    .maxBatchBytes(5 * 1024 * 1024) // 可选：单次批量的最大字节数
    .recordMaxBufferedTime(100) // 可选：记录在内存缓冲的最长时间（毫秒）
    .build();

2. 转换Kafka Message到UserRecord #

KPLFlow仅接收UserRecord类型输入，需添加转换Flow，将Kafka的Message转换为符合要求的UserRecord，同时指定Kinesis的分区键（决定数据落在哪个分片）。

import akka.stream.javadsl.Flow;
import software.amazon.kinesis.producer.UserRecord;
import java.nio.charset.StandardCharsets;

// 定义转换逻辑：Kafka Message → Kinesis UserRecord
Flow<Message, UserRecord, ?> messageToUserRecordFlow = Flow.of(Message.class)
    .map(message -> {
        // 提取Kafka消息的key和value
        String partitionKey = message.key() != null 
            ? new String(message.key(), StandardCharsets.UTF_8) 
            : "default-partition-key"; // 无key时用默认值
        
        byte[] payload = message.value();
        
        // 构建UserRecord
        return UserRecord.builder()
            .streamName("你的目标Kinesis流名称") // 需与settings中的流名称一致
            .partitionKey(partitionKey)
            .data(payload)
            .build();
    });

3. 组装并运行完整流 #

通过KPLFlow.create生成KPL发送Flow，将Kafka Source、转换Flow、KPL Flow串联，最终运行整个流。若需将Flow转为独立Sink，可使用Flow.toMat(Sink.ignore(), Keep.right())，但直接串联更简洁。

import com.github.j5ik2o.akka.kinesis.kpl.KPLFlow;
import akka.stream.javadsl.RunnableGraph;

// 创建KPL发送Flow
Flow<UserRecord, ?, ?> kplFlow = KPLFlow.create(kplFlowSettings, actorSystem);

// 组装流并运行
RunnableGraph<?> runnableGraph = kafkaSource
    .via(messageToUserRecordFlow)
    .via(kplFlow)
    .to(Sink.ignore()); // 忽略KPL返回的RecordMetadata，若需处理可替换为对应Sink

runnableGraph.run(actorSystem);

额外提示 #

• 错误处理：若需处理发送失败的情况，可在KPL Flow后添加重试或异常捕获逻辑，例如：

  .via(kplFlow)
  .retry(3, cause -> cause instanceof RuntimeException)
  

• 流控适配：可调整Kafka Source的消费速率配置，配合KPL的批量参数，避免Kinesis接收压力过大。

内容的提问来源于stack exchange，提问作者Primordium