使用Spring Data Elasticsearch反应式客户端连接AWS OpenSearch报主机不可达错误

使用搭载OpenSearch 1.1.x版本的AWS Elasticsearch服务，在Spring应用中通过spring-data-elasticsearch连接服务，参照官方文档完成Bean配置后出现异常：

1. 本地通过SSH隧道映射远端OpenSearch的443端口到本地9200端口，隧道命令如下：

ssh -4 -i my-creds.pem ec2-user@xxxx.xxxx.xxxx.xxxx -N -L 9200:vpc-my-custom-domain-etc.us-east-1.es.amazonaws.com:443

2. 本地浏览器可通过9200端口正常访问OpenSearch dashboard，使用OpenSearch官方提供的同步RestHighLevelClient、关闭SSL主机名校验后可正常连接服务。
3. 切换为Spring反应式客户端后报如下错误：

reactor.core.Exceptions$ErrorCallbackNotImplemented: org.springframework.data.elasticsearch.client.NoReachableHostException: Host 'localhost:9200' not reachable. Cluster state is offline.
Caused by: org.springframework.data.elasticsearch.client.NoReachableHostException: Host 'localhost:9200' not reachable. Cluster state is offline.

原反应式客户端配置存在两处核心问题：

• 默认使用http协议发起请求，但SSH隧道映射的是远端OpenSearch的https端口，协议不匹配导致连接失败
• 没有配置SSL校验规则，本地访问时证书域名与localhost不匹配，SSL校验失败后判定主机不可达

修改反应式客户端配置，启用https协议并配置适配本地调试的SSL规则即可，修改后配置如下：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
import org.springframework.data.elasticsearch.client.reactive.ReactiveRestClients;
import org.springframework.data.elasticsearch.config.AbstractReactiveElasticsearchConfiguration;
import org.springframework.data.elasticsearch.core.ReactiveElasticsearchOperations;
import org.springframework.data.elasticsearch.core.ReactiveElasticsearchTemplate;
import org.springframework.data.elasticsearch.repository.config.EnableReactiveElasticsearchRepositories;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import reactor.netty.http.client.HttpClient;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.security.cert.X509Certificate;

@Configuration
@EnableReactiveElasticsearchRepositories(basePackages = {"com.elastic.repo"})
public class ElasticRestHighLevelClientConfig extends AbstractReactiveElasticsearchConfiguration {

    @Override
    @Bean
    public ReactiveElasticsearchClient reactiveElasticsearchClient() {
        // 配置信任所有证书的SSL上下文（仅本地开发用，生产环境请替换为合法证书校验逻辑）
        TrustManager[] trustAllCerts = new TrustManager[]{
            new X509TrustManager() {
                public X509Certificate[] getAcceptedIssuers() {
                    return new X509Certificate[0];
                }
                public void checkClientTrusted(X509Certificate[] certs, String authType) {}
                public void checkServerTrusted(X509Certificate[] certs, String authType) {}
            }
        };
        SSLContext sslContext;
        try {
            sslContext = SSLContext.getInstance("TLS");
            sslContext.init(null, trustAllCerts, new java.security.SecureRandom());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        HttpClient httpClient = HttpClient.create()
                .secure(sslContextSpec -> sslContextSpec
                        .sslContext(sslContext)
                        .disableHostnameVerification()); // 关闭主机名校验
        ClientHttpConnector connector = new ReactorClientHttpConnector(httpClient);

        final ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo("localhost:9200")
                .usingSsl() // 启用https协议
                .withClientHttpConnector(connector) // 注入自定义的http连接器
                .build();
        return ReactiveRestClients.create(clientConfiguration);
    }

    @Bean
    public ReactiveElasticsearchOperations elasticsearchOperations(ReactiveElasticsearchClient reactiveElasticsearchClient) {
        return new ReactiveElasticsearchTemplate(reactiveElasticsearchClient);
    }
}

注意：上述配置中信任所有证书、关闭主机名校验的逻辑仅适用于本地开发调试场景，生产环境请替换为正式的SSL证书校验逻辑，避免安全风险。如果AWS OpenSearch服务开启了IAM身份认证，还需要额外配置请求签名拦截器。

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