Connecting to ES with Spring Data Elasticsearch (reactive) gives error host not reachable

Question

I'm running on an aws-elasticsearch (with OpenSearch 1.1.x) service and im trying to connect with it from a spring application using spring-data-elasticsearch, according to the doc i configured the bean as it says.

on my local i used a ssh tunnel from my aws account.

i used this command:

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

so i can connect with OpenSearch dashboard over localhost in my browser through port 9200.

Using the OpenSearch RestHighLevelClient from OpenSearch and disabling the ssl i can connect and it works just fine here the config with OS RHLC:

import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;

public class OSSCLientWorks{

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

    public void request(String indexName, Map<String, Object> doc) throws IOException {

        //Create a client.
        RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "https"))
                .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
                        //.addInterceptorFirst(interceptor) //-> for AwsRequestInterceptor due to some struggles i had, not necessary to work with localhost
                        .setSSLHostnameVerifier((hostname, session) -> true));
        try (RestHighLevelClient hlClient = new RestHighLevelClient(builder)) {

            CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);

            var createIndexResp = hlClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
            log.info("Create index resp {}", createIndexResp);

            IndexRequest indexRequest = new IndexRequest(createIndexResp.index())
                    .id(String.valueOf(doc.get("id")))
                    .source(doc);
            var response = hlClient.index(indexRequest, RequestOptions.DEFAULT);
            var resp = response.toString();
            log.info("response is {}", json);
        }
    }

}

, but when i try with spring and its reactive client i get this error:

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.

here is the config i used to work with spring-data-elasticsearch:

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.EnableElasticsearchRepositories;
import org.springframework.data.elasticsearch.repository.config.EnableReactiveElasticsearchRepositories;


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

    @Override
    @Bean
    public ReactiveElasticsearchClient reactiveElasticsearchClient() {
        final ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo("localhost:9200")
                .build();
        return ReactiveRestClients.create(clientConfiguration);
    }


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

i also tried some solutions other people posted here on SO and Github, but the problem persists, does anybody have a workaround for this? what am i doing wrong?

here i did a demo for the trouble

Thank you very much in advance!

EDIT: clarity

Answer 1

NoReachableHostException is just a generic error they throw when lookupActiveHost(HostProvider interface) fails.

You should debug what happens before - for you it's probably here:

@Override public Mono clusterInfo() {

    return createWebClient(endpoint) //
            .head().uri("/") //
            .exchangeToMono(it -> {
                if (it.statusCode().isError()) {
                    state = ElasticsearchHost.offline(endpoint);
                } else {
                    state = ElasticsearchHost.online(endpoint);
                }
                return Mono.just(state);
            }).onErrorResume(throwable -> {
                state = ElasticsearchHost.offline(endpoint);
                clientProvider.getErrorListener().accept(throwable);
                return Mono.just(state);
            }).map(elasticsearchHost -> new ClusterInformation(Collections.singleton(elasticsearchHost)));
}

see what is the real exception on error resume.

I bet you will get SSL Handshake Exception, you can fix it in the clientConfiguration with .usingSsl({SSL CONTEXT HERE})

You can create insecure context like this(convert to java if needed):

SSLContext.getInstance("TLS") .apply { init(null, InsecureTrustManagerFactory.INSTANCE.trustManagers, SecureRandom()) }

Answer 2

You have to configure to use SSL for the reactive client with one of the usingSsl()methods:

@Override
    @Bean
    public ReactiveElasticsearchClient reactiveElasticsearchClient() {
        final ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo("localhost:9200")
                .usingSsl()                 // <-- 
                .build();
        return ReactiveRestClients.create(clientConfiguration);
    }