新版Elasticsearch Java API Client索引Kafka消息时抛出空指针异常

问题描述 #

从HLRC迁移至新版Elasticsearch Java API Client过程中，整体迁移顺畅，但特定customDto类对应文档始终无法正常索引。
运行环境：

• Spring Boot v2.6.8
• Elasticsearch 7.17.3
  现象：
• 调试确认Kafka传递的payload不为空，单步调试可正常获取doc的id字段
• 执行索引请求逻辑（疑似.build()调用阶段）抛出org.springframework.kafka.listener.ListenerExecutionFailedException
• 无文档成功写入索引，但接口返回200状态码
• 其他独立消费者采用相同逻辑从Kafka消费数据写入其他索引，可正常运行

客户端配置代码：

@Configuration
public class ClientConfiguration{

    @Autowired
    private InternalProperties conf;

    public ElasticsearchClient sslClient(){
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(conf.getElasticsearchUser(), conf.getElasticsearchPassword()));
        HttpHost httpHost = new HttpHost(conf.getElasticsearchAddress(), conf.getElasticsearchPort(), "https");
        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
        try {
            SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, (x509Certificates, s) -> true).build();
            restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    return httpClientBuilder.setSSLContext(sslContext)
                            .setDefaultCredentialsProvider(credentialsProvider);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
        RestClient restClient=restClientBuilder.build();
        
        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());
        
        ElasticsearchClient client = new ElasticsearchClient(transport);
        return client;
    }

}

索引逻辑代码：

@Service
public class ThisDtoIndexClass extends ConfigAndProperties{

    public ThisDtoIndexClass() {

    }

    //client is declared in the class it's extending from 
    public ThisDtoIndexClass(@Autowired ClientConfiguration esClient) {
            this.client = esClient.sslClient();
    }

    @KafkaListener(topics = "esTopic")
    public void in(@Payload(required = false) customDto doc)
        throws ThisDtoIndexClassException, ElasticsearchException, IOException {
        if(doc!= null && doc.getId() != null) {
            IndexRequest.Builder<customDto > indexReqBuilder = new IndexRequest.Builder<>();
            indexReqBuilder.index("index-for-this-Dto");
            indexReqBuilder.id(doc.getId());
            indexReqBuilder.document(doc);

            IndexResponse response = client.index(indexReqBuilder.build());
        } else {
            throw new ThisDtoIndexClassException("document is null");
        }

    }
}

排查方向 #

按优先级从高到低排查：

• 先提取异常根因
  ListenerExecutionFailedException是Spring Kafka对监听器内所有异常的包装类，本身不指向具体问题。在监听器逻辑外层加catch块打印完整异常栈，或者直接断点取异常的getCause()链，绝大多数场景下根因是嵌套的序列化异常、ES端返回的操作错误，不要被外层异常误导。
• 单独验证customDto的序列化逻辑
  其他DTO索引正常，仅该DTO失败，优先排查序列化问题。写单元测试直接用和ES客户端相同的JacksonJsonpMapper实例，把构造好的customDto对象序列化为JSON字符串，观察是否抛出异常。常见问题包括：DTO存在循环引用、特殊类型字段（日期、大数值、自定义枚举、第三方类实例）无对应序列化器、私有字段无标准getter/setter、Jackson序列化配置和DTO注解不匹配。
• 验证ES端索引写入逻辑
  不要只看接口200状态码，200仅代表集群连通、认证通过，不代表索引操作成功。把上一步序列化得到的JSON字符串，直接通过curl或Kibana开发工具调用索引写入接口提交到index-for-this-Dto，查看ES返回的具体响应，常见问题包括：索引mapping和字段类型不匹配、索引处于只读状态、集群磁盘水位线触发写保护、字段值违反mapping设置（比如超过keyword长度限制、分词器配置冲突）。
• 修复客户端配置问题
  当前ClientConfiguration中的sslClient()方法没有加@Bean注解，每次调用都会新建一个RestClient、ElasticsearchTransport、ElasticsearchClient实例，会引发连接泄漏、配置不一致问题。给该方法加上@Bean注解，将ElasticsearchClient注册为Spring单例Bean，统一注入使用，不要每次调用方法新建客户端。同时检查项目依赖树，确认Jackson相关包、ES Java客户端包、JSONP相关包无版本冲突。
• 验证Kafka消息反序列化结果
  即使能正常获取id字段，也不代表整个customDto反序列化正常。在执行索引请求前，打印doc对象的实际类名、全字段值，确认反序列化出来的是原生POJO对象，不是ByteBuddy代理类、半初始化对象，不存在字段缺失、字段值类型异常的问题。

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