使用AWS Lambda Java函数从自托管Kafka集群读取数据及解码问题

问题背景 #

已成功实现AWS官方博客《将自托管Apache Kafka作为AWS Lambda事件源》中的方案，但改用Java编写Lambda代码。当前收到的事件payload里，value字段是Base64编码格式，需要提取并解码该值。相关信息如下：

返回的Payload结构 #

{
    "eventSource": "SelfManagedKafka",
    "bootstrapServers": "sp-dev-broker-0.stre.nsawsdev.mycompany.com:9093",
    "records": {
        "lambda_test-0": [
            {
                "topic": "lambda_test",
                "partition": 0,
                "offset": 0,
                "timestamp": 1660073644797,
                "timestampType": "CREATE_TIME",
                "value": "eyJOYW1lIjogIkFuaXJiYW4gQmlzd2FzIn0=",
                "headers": []
            }
        ]
    }
}

现有Java Lambda代码 #

public class App implements RequestHandler<KafkaEvent, String> {

    private static final Logger log = LoggerFactory.getLogger(App.class);

    @Override
    public String handleRequest(KafkaEvent event, Context context) {

        log.info("Lambda function is invoked:" +event.getRecords().values().toString());
        return null;
    }
}

CloudWatch日志输出 #

Lambda function is invoked:[[KafkaEvent.KafkaEventRecord(topic=lambda_test, partition=0, offset=0, timestamp=1660073644797, timestampType=CREATE_TIME, key=null, value=eyJOYW1lIjogIkFuaXJiYW4gQmlzd2FzIn0=)]]

解决方案 #

步骤1：遍历Kafka事件中的所有记录 #

Lambda收到的KafkaEvent里，records是一个Map结构，键为{topic}-{partition}，值为对应分区的记录列表，需要逐层遍历这些记录。

步骤2：Base64解码value字段 #

使用Java自带的java.util.Base64工具类（Java 8及以上版本支持），将Base64编码的字符串解码为原始字符串。

步骤3：（可选）解析解码后的JSON内容 #

如果解码后的字符串是JSON格式，可以用Jackson等JSON库将其转换为Java对象，方便后续业务处理。

修改后的完整代码 #

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Base64;

public class App implements RequestHandler<KafkaEvent, String> {

    private static final Logger log = LoggerFactory.getLogger(App.class);
    private static final Base64.Decoder BASE64_DECODER = Base64.getDecoder();

    @Override
    public String handleRequest(KafkaEvent event, Context context) {
        // 遍历所有分区的记录集合
        for (KafkaEvent.KafkaEventRecordList recordList : event.getRecords().values()) {
            // 遍历当前分区的每条记录
            for (KafkaEvent.KafkaEventRecord record : recordList) {
                String base64Value = record.getValue();
                if (base64Value != null && !base64Value.isEmpty()) {
                    try {
                        // Base64解码成字节数组，再转UTF-8字符串
                        byte[] decodedBytes = BASE64_DECODER.decode(base64Value);
                        String decodedValue = new String(decodedBytes, "UTF-8");
                        log.info("解码后的value内容: {}", decodedValue);

                        // （可选）如果是JSON格式，解析为Java对象
                        // ObjectMapper objectMapper = new ObjectMapper();
                        // User user = objectMapper.readValue(decodedValue, User.class);
                        // log.info("解析后的用户姓名: {}", user.getName());

                    } catch (IllegalArgumentException e) {
                        log.error("Base64解码失败: {}", e.getMessage());
                    } catch (Exception e) {
                        log.error("处理记录时出错: {}", e.getMessage());
                    }
                }
            }
        }
        return "处理完成";
    }

    // （可选）用于解析JSON的POJO类
    // static class User {
    //     private String Name;
    //     public String getName() { return Name; }
    //     public void setName(String name) { Name = name; }
    // }
}

代码说明 #

• 遍历逻辑：通过event.getRecords().values()获取所有分区的记录列表，再逐个遍历每条消息记录。
• Base64解码：利用Java原生的Base64解码器处理编码字符串，避免引入额外依赖。
• 异常处理：捕获解码失败等异常，防止Lambda函数因无效输入崩溃。
• JSON解析（可选）：若解码后是JSON结构，可引入Jackson依赖，将字符串转为对应Java对象，便于业务逻辑处理。

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