Choosing an Elasticsearch client for Kotlin and Spring Boot

1
rootProject.name = "user-search"
2

3
include("search-index", "client-tour")

1
plugins {
2
    kotlin("jvm") version "2.3.21" apply false
3
    kotlin("plugin.spring") version "2.3.21" apply false
4
    id("org.springframework.boot") version "4.1.1" apply false
5
}
6

7
subprojects {
8
    group = "in.o612.eng.usersearch"
9
    version = "0.1.0"
10
    repositories { mavenCentral() }
11
}

1
import org.springframework.boot.gradle.plugin.SpringBootPlugin
2

3
plugins {
4
    kotlin("jvm")
5
}
6

7
kotlin { jvmToolchain(21) }
8

9
dependencies {
10
    implementation(platform(SpringBootPlugin.BOM_COORDINATES))
11
    api("co.elastic.clients:elasticsearch-java")
12
    implementation("tools.jackson.module:jackson-module-kotlin")
13
}

1
package `in`.o612.eng.usersearch.index
2

3
import java.time.Instant
4
import java.time.LocalDate
5

6
/** One profile as stored in the user-profile index. Field names match the mapping. */
7
data class UserProfileDocument(
8
    val userId: String,
9
    val fullName: String,
10
    val firstName: String,
11
    val lastName: String,
12
    val email: String,
13
    val mobileNumber: String,
14
    val city: String,
15
    val state: String,
16
    val country: String,
17
    val pincode: String,
18
    val dateOfBirth: LocalDate?,
19
    val gender: String,
20
    val accountStatus: String,
21
    val createdAt: Instant,
22
    val updatedAt: Instant,
23
)

1
package `in`.o612.eng.usersearch.index
2

3
/** Names and definitions of the versioned user-profile index. */
4
object UserProfileIndex {
5
    const val READ_ALIAS = "user-profile-read"
6
    const val WRITE_ALIAS = "user-profile-write"
7

8
    fun indexName(version: Int) = "user-profile-v$version"
9

10
    /** The reviewed index definition for [version], from `es/user-profile-v<version>.json`. */
11
    fun definition(version: Int): String {
12
        val path = "/es/user-profile-v$version.json"
13
        val stream = UserProfileIndex::class.java.getResourceAsStream(path)
14
            ?: error("No index definition at $path")
15
        return stream.use { it.readBytes().decodeToString() }
16
    }
17
}

1
package `in`.o612.eng.usersearch.index
2

3
import co.elastic.clients.json.JsonpMapper
4
import co.elastic.clients.json.jackson.Jackson3JsonpMapper
5
import com.fasterxml.jackson.annotation.JsonInclude
6
import tools.jackson.databind.json.JsonMapper
7
import tools.jackson.module.kotlin.KotlinModule
8

9
/**
10
 * The JSON mapper for the Elasticsearch client.
11
 *
12
 * Spring Boot's default, `Jackson3JsonpMapper()`, registers no modules, so it cannot
13
 * construct Kotlin data classes. This one adds the Kotlin module and keeps the
14
 * client's default of omitting null properties.
15
 */
16
object ElasticsearchJson {
17
    fun jsonpMapper(): JsonpMapper = Jackson3JsonpMapper(
18
        JsonMapper.builder()
19
            .addModule(KotlinModule.Builder().build())
20
            .changeDefaultPropertyInclusion { it.withValueInclusion(JsonInclude.Include.NON_NULL) }
21
            .build(),
22
    )
23
}

1
import org.springframework.boot.gradle.plugin.SpringBootPlugin
2

3
plugins {
4
    kotlin("jvm")
5
    kotlin("plugin.spring")
6
    id("org.springframework.boot")
7
}
8

9
kotlin { jvmToolchain(21) }
10

11
dependencies {
12
    implementation(platform(SpringBootPlugin.BOM_COORDINATES))
13
    implementation(project(":search-index"))
14
    implementation("org.springframework.boot:spring-boot-starter-elasticsearch")
15
    implementation("org.springframework.boot:spring-boot-starter-data-elasticsearch")
16
    implementation("tools.jackson.module:jackson-module-kotlin")
17
    implementation("org.jetbrains.kotlin:kotlin-reflect")
18
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core")
19
}

1
spring:
2
  main:
3
    web-application-type: none
4
  elasticsearch:
5
    uris: ${ELASTICSEARCH_URIS:http://localhost:9200}
6
    username: ${ELASTICSEARCH_USERNAME}
7
    password: ${ELASTICSEARCH_PASSWORD}

1
package `in`.o612.eng.usersearch.tour
2

3
import org.springframework.boot.SpringApplication
4
import org.springframework.boot.autoconfigure.SpringBootApplication
5
import org.springframework.boot.runApplication
6
import kotlin.system.exitProcess
7

8
@SpringBootApplication
9
class ClientTourApplication
10

11
fun main(args: Array<String>) {
12
    // Run every tour step, then close the context so the client's I/O threads stop.
13
    val context = runApplication<ClientTourApplication>(*args)
14
    exitProcess(SpringApplication.exit(context))
15
}

1
package `in`.o612.eng.usersearch.tour
2

3
import co.elastic.clients.json.JsonpMapper
4
import `in`.o612.eng.usersearch.index.ElasticsearchJson
5
import org.springframework.context.annotation.Bean
6
import org.springframework.context.annotation.Configuration
7

8
@Configuration
9
class ElasticsearchConfig {
10
    @Bean
11
    fun jsonpMapper(): JsonpMapper = ElasticsearchJson.jsonpMapper()
12
}

1
package `in`.o612.eng.usersearch.tour
2

3
import co.elastic.clients.elasticsearch.ElasticsearchClient
4
import co.elastic.clients.elasticsearch._types.FieldValue
5
import co.elastic.clients.elasticsearch._types.query_dsl.Operator
6
import co.elastic.clients.elasticsearch.core.SearchRequest
7
import `in`.o612.eng.usersearch.index.UserProfileDocument
8
import `in`.o612.eng.usersearch.index.UserProfileIndex
9
import org.springframework.boot.ApplicationRunner
10
import org.springframework.context.annotation.Bean
11
import org.springframework.context.annotation.Configuration
12
import org.springframework.core.annotation.Order
13
import java.io.StringReader
14

15
@Configuration
16
class TypedClientTour {
17

18
    @Bean
19
    @Order(1)
20
    fun typedClient(client: ElasticsearchClient) = ApplicationRunner {
21
        println("JSON mapper: ${client._jsonpMapper().javaClass.simpleName}")
22

23
        // Get by id: one shard, real time, deserialised into a Kotlin data class.
24
        val profile = client.get(
25
            { it.index(UserProfileIndex.READ_ALIAS).id("42") },
26
            UserProfileDocument::class.java,
27
        ).source()
28
        println("get 42 -> ${profile?.fullName}, ${profile?.city}, ${profile?.accountStatus}")
29

30
        // Search built with the typed DSL.
31
        val response = client.search({ s ->
32
            s.index(UserProfileIndex.READ_ALIAS)
33
                .size(3)
34
                .source { src -> src.filter { f -> f.includes("userId", "fullName", "city") } }
35
                .query { q ->
36
                    q.bool { b ->
37
                        b.must { m ->
38
                            m.match { mt -> mt.field("fullName.prefix").query("prashant ku").operator(Operator.And) }
39
                        }.filter { f ->
40
                            f.term { t -> t.field("state").value(FieldValue.of("bihar")) }
41
                        }.filter { f ->
42
                            f.term { t -> t.field("accountStatus").value(FieldValue.of("ACTIVE")) }
43
                        }
44
                    }
45
                }
46
        }, ProfileSummary::class.java)
47
        println("typed search -> total ${response.hits().total()?.value()}, first ${response.hits().hits().map { it.source()?.fullName }}")
48

49
        // The same request from JSON, for example pasted from Kibana Dev Tools.
50
        val json = """
51
            { "size": 0,
52
              "query": { "bool": { "filter": [
53
                { "term": { "state": "bihar" } },
54
                { "term": { "accountStatus": "ACTIVE" } } ] } } }
55
        """.trimIndent()
56
        val fromJson = SearchRequest.of { it.index(UserProfileIndex.READ_ALIAS).withJson(StringReader(json)) }
57
        val total = client.search(fromJson, Void::class.java).hits().total()
58
        println("withJson search -> total ${total?.value()} (${total?.relation()})")
59

60
        // Ask for an exact count when the number matters.
61
        val exact = client.search({ s ->
62
            s.index(UserProfileIndex.READ_ALIAS).size(0).trackTotalHits { t -> t.enabled(true) }
63
                .query { q -> q.term { t -> t.field("state").value(FieldValue.of("bihar")) } }
64
        }, Void::class.java).hits().total()
65
        println("exact total -> ${exact?.value()} (${exact?.relation()})")
66
    }
67
}
68

69
/** Only the fields a result list needs. */
70
data class ProfileSummary(val userId: String, val fullName: String, val city: String)

1
package `in`.o612.eng.usersearch.tour
2

3
import co.elastic.clients.transport.rest5_client.low_level.Request
4
import co.elastic.clients.transport.rest5_client.low_level.Rest5Client
5
import org.springframework.boot.ApplicationRunner
6
import org.springframework.context.annotation.Bean
7
import org.springframework.context.annotation.Configuration
8
import org.springframework.core.annotation.Order
9

10
@Configuration
11
class LowLevelClientTour {
12

13
    @Bean
14
    @Order(2)
15
    fun lowLevelClient(restClient: Rest5Client) = ApplicationRunner {
16
        val request = Request("GET", "/_cat/aliases/user-profile-*").apply {
17
            addParameter("v", "true")
18
            addParameter("h", "alias,index,is_write_index")
19
        }
20
        val response = restClient.performRequest(request)
21
        println("low-level status ${response.statusCode}")
22
        print(response.entity.content.readAllBytes().decodeToString())
23
    }
24
}

1
package `in`.o612.eng.usersearch.tour
2

3
import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient
4
import co.elastic.clients.elasticsearch.ElasticsearchClient
5
import co.elastic.clients.elasticsearch._types.FieldValue
6
import `in`.o612.eng.usersearch.index.UserProfileIndex
7
import kotlinx.coroutines.async
8
import kotlinx.coroutines.coroutineScope
9
import kotlinx.coroutines.future.await
10
import kotlinx.coroutines.runBlocking
11
import org.springframework.boot.ApplicationRunner
12
import org.springframework.context.annotation.Bean
13
import org.springframework.context.annotation.Configuration
14
import org.springframework.core.annotation.Order
15

16
@Configuration
17
class AsyncClientTour {
18

19
    @Bean
20
    @Order(3)
21
    fun asyncClient(client: ElasticsearchClient) = ApplicationRunner {
22
        // Shares the blocking client's transport: one connection pool for both.
23
        val async = ElasticsearchAsyncClient(client._transport())
24

25
        suspend fun countIn(state: String): Long =
26
            async.count { c ->
27
                c.index(UserProfileIndex.READ_ALIAS)
28
                    .query { q -> q.term { t -> t.field("state").value(FieldValue.of(state)) } }
29
            }.await().count()
30

31
        runBlocking {
32
            coroutineScope {
33
                val states = listOf("bihar", "kerala", "karnataka")
34
                val counts = states.map { state -> async { state to countIn(state) } }.map { it.await() }
35
                println("async counts -> $counts")
36
            }
37
        }
38
    }
39
}

1
package `in`.o612.eng.usersearch.tour
2

3
import co.elastic.clients.elasticsearch._types.FieldValue
4
import `in`.o612.eng.usersearch.index.UserProfileIndex
5
import org.springframework.boot.ApplicationRunner
6
import org.springframework.context.annotation.Bean
7
import org.springframework.context.annotation.Configuration
8
import org.springframework.core.annotation.Order
9
import org.springframework.data.annotation.Id
10
import org.springframework.data.domain.Page
11
import org.springframework.data.domain.PageRequest
12
import org.springframework.data.domain.Pageable
13
import org.springframework.data.domain.Sort
14
import org.springframework.data.elasticsearch.annotations.DateFormat
15
import org.springframework.data.elasticsearch.annotations.Document
16
import org.springframework.data.elasticsearch.annotations.Field
17
import org.springframework.data.elasticsearch.annotations.FieldType
18
import org.springframework.data.elasticsearch.client.elc.NativeQuery
19
import org.springframework.data.elasticsearch.core.ElasticsearchOperations
20
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates
21
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository
22
import java.time.Instant
23

24
/** The profile as Spring Data sees it. createIndex = false: the index is managed from reviewed JSON. */
25
@Document(indexName = UserProfileIndex.READ_ALIAS, createIndex = false)
26
data class ProfileEntity(
27
    @Id val userId: String,
28
    val fullName: String,
29
    val city: String,
30
    val state: String,
31
    val accountStatus: String,
32
    @Field(type = FieldType.Date, format = [DateFormat.date_time_no_millis])
33
    val updatedAt: Instant,
34
)
35

36
interface ProfileRepository : ElasticsearchRepository<ProfileEntity, String> {
37
    fun findByCityAndAccountStatus(city: String, accountStatus: String, pageable: Pageable): Page<ProfileEntity>
38
}
39

40
@Configuration
41
class SpringDataTour {
42

43
    @Bean
44
    @Order(4)
45
    fun springData(repository: ProfileRepository, operations: ElasticsearchOperations) = ApplicationRunner {
46
        // A derived query: Spring Data writes the query from the method name.
47
        val page = repository.findByCityAndAccountStatus(
48
            "patna", "ACTIVE", PageRequest.of(0, 3, Sort.by(Sort.Direction.DESC, "updatedAt")),
49
        )
50
        println("repository -> total ${page.totalElements}, first ${page.content.map { "${it.fullName} ${it.updatedAt}" }}")
51

52
        // NativeQuery: Spring Data's paging and mapping around the Java client's query builder.
53
        val query = NativeQuery.builder()
54
            .withQuery { q ->
55
                q.bool { b ->
56
                    b.must { m -> m.match { mt -> mt.field("fullName").query("prashant kumar") } }
57
                        .filter { f -> f.term { t -> t.field("state").value(FieldValue.of("bihar")) } }
58
                }
59
            }
60
            .withPageable(PageRequest.of(0, 3))
61
            .withTrackTotalHits(true)
62
            .build()
63
        val hits = operations.search(query, ProfileEntity::class.java, IndexCoordinates.of(UserProfileIndex.READ_ALIAS))
64
        println("operations -> total ${hits.totalHits}, scores ${hits.searchHits.map { it.score }}")
65
    }
66
}