On this page

Opensearch

1 Introduction

Micronaut OpenSearch simplifies integration with OpenSearch.

OpenSearch is the flexible, scalable, open-source way to build solutions for data-intensive applications.

OpenSearchClient

Once you have integrated Micronaut OpenSearch, you are able to inject a bean of type org.opensearch.client.opensearch.OpenSearchClient or org.opensearch.client.opensearch.OpenSearchAsyncClient.

package micronaut.example.service;

import java.util.Iterator;

import jakarta.inject.Singleton;
import micronaut.example.configuration.AppConfiguration;
import micronaut.example.exception.MovieServiceException;

import org.opensearch.client.opensearch.OpenSearchClient;
import org.opensearch.client.opensearch.core.IndexRequest;
import org.opensearch.client.opensearch.core.IndexResponse;
import org.opensearch.client.opensearch.core.SearchResponse;
import org.opensearch.client.opensearch.core.search.Hit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Singleton
class MovieServiceImpl implements MovieService {
    private static final Logger LOG = LoggerFactory.getLogger(MovieServiceImpl.class);
    private final AppConfiguration appConfiguration;
    private final OpenSearchClient client;

    public MovieServiceImpl(AppConfiguration appConfiguration, OpenSearchClient client) {
        this.appConfiguration = appConfiguration;
        this.client = client;
    }

    @Override
    public String saveMovie(Movie movie) {
        try {
            IndexRequest<Movie> indexRequest = createIndexRequest(movie);
            IndexResponse indexResponse = client.index(indexRequest);
            String id = indexResponse.id();
            LOG.info("Document for '{}' {} successfully in ES. The id is: {}", movie, indexResponse.result(), id);
            return id;
        } catch (Exception e) {
            String errorMessage = String.format("An exception occurred while indexing '%s'", movie);
            LOG.error(errorMessage);
            throw new MovieServiceException(errorMessage, e);
        }
    }

    private IndexRequest<Movie> createIndexRequest(Movie movie) {
        return new IndexRequest.Builder<Movie>()
            .index(appConfiguration.getMoviesIndexName())
            .document(movie)
            .build();
    }

    @Override
    public Movie searchMovies(String title) {
        try {
            SearchResponse<Movie> searchResponse = client.search(s ->
                    s.index(appConfiguration.getMoviesIndexName()).query(q ->
                            q.match(m ->
                                    m.field("title").query(fq -> fq.stringValue(title)))), Movie.class);
            LOG.info("Searching for '{}' took {} and found {}",
                    title,
                    searchResponse.took(),
                    searchResponse.hits().total().value());
            Iterator<Hit<Movie>> hits = searchResponse.hits().hits().iterator();
            if (hits.hasNext()) {
                return hits.next().source();
            }
            return null;
        } catch (Exception e) {
            String errorMessage = String.format("An exception occurred while searching for title '%s'", title);
            LOG.error(errorMessage);
            throw new MovieServiceException(errorMessage, e);
        }
    }
}

import logging

from jakarta.inject import Singleton
from java.lang import Exception as JavaException
from org.opensearch.client.opensearch import OpenSearchClient
from org.opensearch.client.opensearch.core import IndexRequest

from micronaut.example.configuration.AppConfiguration import AppConfiguration
from micronaut.example.exception.MovieServiceException import MovieServiceException
from micronaut.example.service.Movie import Movie
from micronaut.example.service.MovieService import MovieService

LOG = logging.getLogger(__name__)


@Singleton
class MovieServiceImpl(MovieService):

    def __init__(self, app_configuration: AppConfiguration, client: OpenSearchClient):
        self.app_configuration = app_configuration
        self.client = client

    def save_movie(self, movie: Movie) -> str:
        try:
            index_request = self.create_index_request(movie)
            index_response = self.client.index(index_request)
            id = index_response.id()
            LOG.info("Document for '%s' %s successfully in ES. The id is: %s", movie, index_response.result(), id)
            return id
        except JavaException as e:
            error_message = f"An exception occurred while indexing '{movie}'"
            LOG.error(error_message)
            raise MovieServiceException(error_message, e)

    def create_index_request(self, movie: Movie) -> IndexRequest[Movie]:
        return (IndexRequest.Builder()
                .index(self.app_configuration.movies_index_name)
                .document(movie)
                .build())

    def search_movies(self, title: str) -> Movie | None:
        try:
            search_response = self.client.search(
                lambda s: s.index(self.app_configuration.movies_index_name).query(
                    lambda q: q.match(
                        lambda m: m.field("title").query(lambda fq: fq.stringValue(title)))),
                Movie)
            LOG.info("Searching for '%s' took %s and found %s",
                     title,
                     search_response.took(),
                     search_response.hits().total().value())
            hits = search_response.hits().hits()
            if not hits.isEmpty():
                return hits.get(0).source()
            return None
        except JavaException as e:
            error_message = f"An exception occurred while searching for title '{title}'"
            LOG.error(error_message)
            raise MovieServiceException(error_message, e)

package micronaut.example.service

import jakarta.inject.Singleton
import micronaut.example.configuration.AppConfiguration
import micronaut.example.exception.MovieServiceException
import org.opensearch.client.opensearch.OpenSearchClient
import org.opensearch.client.opensearch.core.IndexRequest
import org.opensearch.client.opensearch.core.IndexResponse
import org.opensearch.client.opensearch.core.SearchResponse
import org.opensearch.client.opensearch.core.search.Hit
import org.slf4j.Logger
import org.slf4j.LoggerFactory

@Singleton
class MovieServiceImpl(
    private val appConfiguration: AppConfiguration,
    private val client: OpenSearchClient
) : MovieService {

    override fun saveMovie(movie: Movie): String {
        try {
            val indexRequest: IndexRequest<Movie> = createIndexRequest(movie)
            val indexResponse: IndexResponse = client.index(indexRequest)
            val id: String = indexResponse.id()
            LOG.info("Document for '{}' {} successfully in ES. The id is: {}", movie, indexResponse.result(), id)
            return id
        } catch (e: Exception) {
            val errorMessage = String.format("An exception occurred while indexing '%s'", movie)
            LOG.error(errorMessage)
            throw MovieServiceException(errorMessage, e)
        }
    }

    private fun createIndexRequest(movie: Movie): IndexRequest<Movie> {
        return IndexRequest.Builder<Movie>()
            .index(appConfiguration.getMoviesIndexName())
            .document(movie)
            .build()
    }

    override fun searchMovies(title: String): Movie? {
        try {
            val searchResponse: SearchResponse<Movie> = client.search({ s ->
                    s.index(appConfiguration.getMoviesIndexName()).query { q ->
                            q.match { m ->
                                m.field("title").query { fq -> fq.stringValue(title) }
                            }}}, Movie::class.java)
            LOG.info(
                "Searching for '{}' took {} and found {}",
                title,
                searchResponse.took(),
                searchResponse.hits().total()?.value()
            )

            val hits: Iterator<Hit<Movie>> = searchResponse.hits().hits().iterator()
            if (hits.hasNext()) {
                return hits.next().source()
            }
            return null
        } catch (e: Exception) {
            val errorMessage = String.format("An exception occurred while searching for title '%s'", title)
            LOG.error(errorMessage)
            throw MovieServiceException(errorMessage, e)
        }
    }

    companion object {
        private val LOG: Logger = LoggerFactory.getLogger(MovieServiceImpl::class.java)
    }
}

package micronaut.example.service

import micronaut.example.configuration.AppConfiguration
import micronaut.example.exception.MovieServiceException
import jakarta.inject.Singleton
import org.opensearch.client.opensearch.OpenSearchClient
import org.opensearch.client.opensearch.core.IndexRequest
import org.opensearch.client.opensearch.core.IndexResponse
import org.opensearch.client.opensearch.core.SearchResponse
import org.opensearch.client.opensearch.core.search.Hit
import org.slf4j.Logger
import org.slf4j.LoggerFactory

@Singleton
class MovieServiceImpl implements MovieService {

    private static final Logger LOG = LoggerFactory.getLogger(MovieServiceImpl.class)

    private final AppConfiguration appConfiguration

    private final OpenSearchClient client

    MovieServiceImpl(AppConfiguration appConfiguration, OpenSearchClient client) {
        this.appConfiguration = appConfiguration
        this.client = client
    }

    @Override
    String saveMovie(Movie movie) {
        try {
            IndexRequest<Movie> indexRequest = createIndexRequest(movie)

            IndexResponse indexResponse = client.index(indexRequest)
            String id = indexResponse.id()

            LOG.info("Document for '{}' {} successfully in ES. The id is: {}", movie, indexResponse.result(), id)
            return id
        } catch (Exception e) {
            String errorMessage = String.format("An exception occurred while indexing '%s'", movie)
            LOG.error(errorMessage)
            throw new MovieServiceException(errorMessage, e)
        }
    }

    private IndexRequest<Movie> createIndexRequest(Movie movie) {
        return new IndexRequest.Builder<Movie>()
            .index(appConfiguration.getMoviesIndexName())
            .document(movie)
            .build()
    }

    @Override
    Movie searchMovies(String title) {
        try {
            SearchResponse<Movie> searchResponse = client.search((s) ->
                s.index(appConfiguration.getMoviesIndexName())
                    .query(q -> q.match(m ->
                        m.field("title")
                         .query(fq -> fq.stringValue(title))
                    )), Movie.class
            )
            LOG.info("Searching for '{}' took {} and found {}", title, searchResponse.took(), searchResponse.hits().total().value())

            Iterator<Hit<Movie>> hits = searchResponse.hits().hits().iterator()
            if (hits.hasNext()) {
                return hits.next().source()
            }
            return null

        } catch (Exception e) {
            String errorMessage = String.format("An exception occurred while searching for title '%s'", title)
            LOG.error(errorMessage)
            throw new MovieServiceException(errorMessage, e)
        }
    }
}

2 OpenSearch Amazon

To use Micronaut OpenSearch and connect to Amazon OpenSearch Service add the following dependency:

implementation("io.micronaut.opensearch:micronaut-opensearch-amazon")

<dependency>
    <groupId>io.micronaut.opensearch</groupId>
    <artifactId>micronaut-opensearch-amazon</artifactId>
</dependency>

You can configure the connection with:

Moreover, you can create a BeanCreatedEventListener for a bean of type org.opensearch.client.transport.aws.AwsSdk2TransportOptions.Builder, to configure the connection according to your use case.

import io.micronaut.context.event.BeanCreatedEvent;
import io.micronaut.context.event.BeanCreatedEventListener;
import jakarta.inject.Singleton;
import org.opensearch.client.transport.aws.AwsSdk2TransportOptions;

@Singleton
class AwsSdk2TransportOptionsBeanCreatedEventListener implements BeanCreatedEventListener<AwsSdk2TransportOptions.Builder> {

    @Override
    public AwsSdk2TransportOptions.Builder onCreated(BeanCreatedEvent<AwsSdk2TransportOptions.Builder> event) {
        AwsSdk2TransportOptions.Builder builder = event.getBean();

        // Modify the builder here
        return builder;
    }

}

NOTE: Empty tag `endclazz` in `test-suite-python/src/test/python/micronaut/example/aws/AwsSdk2TransportOptionsBeanCreatedEventListener.py`.

import io.micronaut.context.event.BeanCreatedEvent
import io.micronaut.context.event.BeanCreatedEventListener
import jakarta.inject.Singleton
import org.opensearch.client.transport.aws.AwsSdk2TransportOptions

@Singleton
class AwsSdk2TransportOptionsBeanCreatedEventListener : BeanCreatedEventListener<AwsSdk2TransportOptions.Builder> {

    override fun onCreated(event: BeanCreatedEvent<AwsSdk2TransportOptions.Builder>): AwsSdk2TransportOptions.Builder {
        val builder = event.bean

        // Modify the builder here
        return builder
    }

}

import io.micronaut.context.event.BeanCreatedEvent
import io.micronaut.context.event.BeanCreatedEventListener
import jakarta.inject.Singleton
import org.opensearch.client.transport.aws.AwsSdk2TransportOptions

@Singleton
class AwsSdk2TransportOptionsBeanCreatedEventListener implements BeanCreatedEventListener<AwsSdk2TransportOptions.Builder> {

    @Override
    AwsSdk2TransportOptions.Builder onCreated(BeanCreatedEvent<AwsSdk2TransportOptions.Builder> event) {
        AwsSdk2TransportOptions.Builder builder = event.bean

        // Modify the builder here
        return builder
    }

}

3 OpenSearch using Apache HttpClient 5 Transport

To use Micronaut OpenSearch and connect using Apache HttpClient 5 transport, add the following dependencies:

implementation("io.micronaut.opensearch:micronaut-opensearch-httpclient5")

<dependency>
    <groupId>io.micronaut.opensearch</groupId>
    <artifactId>micronaut-opensearch-httpclient5</artifactId>
</dependency>

You can configure the connection with:

You can create a BeanCreatedEventListener for a bean of type org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportBuilder to configure the connection according to your use case.

4 OpenSearch RestClient

To use Micronaut OpenSearch and connect with RestClient-based transport add the following dependencies:

implementation("io.micronaut.opensearch:micronaut-opensearch-restclient")

<dependency>
    <groupId>io.micronaut.opensearch</groupId>
    <artifactId>micronaut-opensearch-restclient</artifactId>
</dependency>

implementation("io.micronaut:micronaut-jackson-databind")

<dependency>
    <groupId>io.micronaut</groupId>
    <artifactId>micronaut-jackson-databind</artifactId>
</dependency>

You can configure the connection with:

You can create BeanCreatedEventListeners for a beans of type org.apache.http.client.config.RequestConfig.Builder, org.apache.http.impl.nio.client.HttpAsyncClientBuilder, and org.opensearch.client.RestClientBuilder to configure the connection according to your use case.

5 Health

After you add the management dependency, the Health Endpoint exposes a health indicator for OpenSearch.

You can disable it with:

6 Release History

For this project, you can find a list of releases (with release notes) here:

https://github.com/micronaut-projects/micronaut-opensearch/releases

7 Repository

You can find the source code of this project in this repository:

https://github.com/micronaut-projects/micronaut-opensearch