Part 14 of 1878% complete

2026-05-26•13 min read

Backfill, the outbox relay, and zero-downtime reindexing

Until now, the index was filled by a shell script, and the outbox from chapter 04 has been collecting events that nothing consumed. This chapter builds the component that owns every write to Elasticsearch: a Kotlin indexer application with two jobs. The backfill copies every profile from PostgreSQL into a named index, reading with keyset pagination and writing through BulkIngester with backpressure and per-item error handling. The relay drains the outbox continuously, re-reading current rows as chapter 04 designed.

Then you use both to move the live system from user-profile-v1 to a new user-profile-v2, with changes made in PostgreSQL while the backfill runs, and without the Search API noticing: create, backfill, catch up, validate, swap both aliases atomically, and keep v1 current for a rollback.

You need the lab and the user-search project from chapter 13. The chapter takes about 90 minutes.

Why mapping changes mean a new index

Chapter 02 showed that an existing field’s type cannot change, and chapter 06 that analysis settings apply to terms already written. The same holds for most mapping decisions that matter: turning off doc values, changing an analyser, splitting a field into sub-fields for existing documents, or changing the primary shard count. Adding a brand-new field is allowed in place, but existing documents do not gain values for it until they are indexed again. In practice, any change to how existing profiles are indexed means writing every document again, into a new index. The versioned names and separate aliases from chapter 05 exist for this moment.

The v2 index in this chapter makes two changes found in earlier chapters:

Display sub-fields for facets. Chapter 12’s facets returned normalised keys such as maharashtra. city.display and state.display are keywords without a normaliser, so facets can show Maharashtra.
No doc values on email and mobileNumber. Chapter 08’s disk-usage breakdown showed email as the largest indexed field, partly because of doc values that no query shape uses. Exact lookups still work: they use the inverted index, not doc values.

Create es/user-profile-v2.json in the lab directory, and copy it to search-index/src/main/resources/es/ in the project. It is v1 with those changes, and without an aliases block, because the aliases move to v2 only at the swap:

1
{
2
  "settings": {
3
    "index": {
4
      "number_of_shards": 1,
5
      "number_of_replicas": 0,
6
      "refresh_interval": "1s"
7
    },
8
    "analysis": {
9
      "filter": {
10
        "name_edge_ngram": {
11
          "type": "edge_ngram",
12
          "min_gram": 2,
13
          "max_gram": 15
14
        },
15
        "name_synonyms": {
16
          "type": "synonym_graph",
17
          "synonyms_set": "profile-name-synonyms",
18
          "updateable": true
19
        }
20
      },
21
      "analyzer": {
22
        "name_index": {
23
          "type": "custom",
24
          "tokenizer": "standard",
25
          "filter": ["lowercase", "asciifolding"]
26
        },
27
        "name_search": {
28
          "type": "custom",
29
          "tokenizer": "standard",
30
          "filter": ["lowercase", "asciifolding", "name_synonyms"]
31
        },
32
        "name_prefix_index": {
33
          "type": "custom",
34
          "tokenizer": "standard",
35
          "filter": ["lowercase", "asciifolding", "name_edge_ngram"]
36
        }
37
      },
38
      "normalizer": {
39
        "lowercase_ascii": {
40
          "type": "custom",
41
          "filter": ["lowercase", "asciifolding"]
42
        },
43
        "lowercase_only": {
44
          "type": "custom",
45
          "filter": ["lowercase"]
46
        }
47
      }
48
    }
49
  },
50
  "mappings": {
51
    "dynamic": "strict",
52
    "properties": {
53
      "userId":        { "type": "keyword" },
54
      "fullName": {
55
        "type": "text",
56
        "analyzer": "name_index",
57
        "search_analyzer": "name_search",
58
        "fields": {
59
          "prefix":  { "type": "text", "analyzer": "name_prefix_index", "search_analyzer": "name_index" },
60
          "keyword": { "type": "keyword", "normalizer": "lowercase_ascii", "ignore_above": 256 }
61
        }
62
      },
63
      "firstName":     { "type": "text", "analyzer": "name_index", "search_analyzer": "name_search" },
64
      "lastName":      { "type": "text", "analyzer": "name_index", "search_analyzer": "name_search" },
65
      "email":         { "type": "keyword", "normalizer": "lowercase_only", "doc_values": false },
66
      "mobileNumber":  { "type": "keyword", "doc_values": false },
67
      "city": {
68
        "type": "keyword",
69
        "normalizer": "lowercase_ascii",
70
        "fields": {
71
          "text":    { "type": "text", "analyzer": "name_index" },
72
          "display": { "type": "keyword" }
73
        }
74
      },
75
      "state": {
76
        "type": "keyword",
77
        "normalizer": "lowercase_ascii",
78
        "fields": {
79
          "text":    { "type": "text", "analyzer": "name_index" },
80
          "display": { "type": "keyword" }
81
        }
82
      },
83
      "country":       { "type": "keyword" },
84
      "pincode":       { "type": "keyword" },
85
      "dateOfBirth":   { "type": "date", "index": false, "doc_values": false },
86
      "gender":        { "type": "keyword" },
87
      "accountStatus": { "type": "keyword" },
88
      "createdAt":     { "type": "date" },
89
      "updatedAt":     { "type": "date" }
90
    }
91
  }
92
}

Compared with v1, only three blocks differ: email and mobileNumber gain "doc_values": false, and city and state each gain a display sub-field.

Stage 1 — The indexer module

Add the module to settings.gradle.kts:

1
rootProject.name = "user-search"
2

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

The indexer needs JDBC for PostgreSQL and the Elasticsearch client, and nothing for the web:

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 {
10
    jvmToolchain(21)
11
    compilerOptions { freeCompilerArgs.add("-Xjsr305=strict") }
12
}
13

14
dependencies {
15
    implementation(platform(SpringBootPlugin.BOM_COORDINATES))
16
    implementation(project(":search-index"))
17
    implementation("org.springframework.boot:spring-boot-starter-elasticsearch")
18
    implementation("org.springframework.boot:spring-boot-starter-jdbc")
19
    implementation("tools.jackson.module:jackson-module-kotlin")
20
    implementation("org.jetbrains.kotlin:kotlin-reflect")
21
    runtimeOnly("org.postgresql:postgresql")
22
}

1
spring:
2
  application:
3
    name: indexer
4
  main:
5
    web-application-type: none
6
  datasource:
7
    url: ${DATABASE_URL}
8
    username: ${DATABASE_USERNAME}
9
    password: ${DATABASE_PASSWORD}
10
  elasticsearch:
11
    uris: ${ELASTICSEARCH_URIS}
12
    username: ${ELASTICSEARCH_USERNAME}
13
    password: ${ELASTICSEARCH_PASSWORD}
14
    connection-timeout: 2s
15
    socket-timeout: 60s
16
    restclient:
17
      ssl:
18
        bundle: elasticsearch
19
  ssl:
20
    bundle:
21
      pem:
22
        elasticsearch:
23
          truststore:
24
            certificate: ${ELASTICSEARCH_CA_CERT}
25

26
indexer:
27
  # relay (default, runs until stopped), create, backfill, validate, or swap
28
  command: ${INDEXER_COMMAND:relay}
29
  source-index: ${INDEXER_SOURCE_INDEX:}
30
  target-index: ${INDEXER_TARGET_INDEX:}
31
  relay:
32
    poll-interval: 1s
33
    batch-size: 500
34
    # Indices or aliases every change is written to. Add the new index version during a reindex.
35
    targets: ${INDEXER_RELAY_TARGETS:user-profile-write}
36
  backfill:
37
    page-size: 2000
38
    bulk-operations: 1000
39
    concurrent-requests: 2

The socket timeout is longer than the Search API’s, because a bulk request with a thousand documents legitimately takes longer than a search. indexer.command selects what the application does; relay, the default, runs until stopped, and the others run once and exit.

Add a table for permanent failures to the lab database:

1
-- Changes the indexer could not apply and will not retry. Replay: fix the cause, then
2
-- INSERT INTO profile_search_outbox (user_id, row_version, operation) for the affected user_id.
3
CREATE TABLE profile_search_dead_letter (
4
    id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
5
    user_id      bigint      NOT NULL,
6
    row_version  bigint      NOT NULL,
7
    target       text        NOT NULL,
8
    error_type   text        NOT NULL,
9
    error_reason text,
10
    created_at   timestamptz NOT NULL DEFAULT now()
11
);

1
docker compose exec -T postgres psql -v ON_ERROR_STOP=1 -U profiles -d profiles < db/sync/02-dead-letter.sql

The shared IndexAdmin gains settings updates, counts, refresh, and the alias swap. Replace the file:

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

3
import co.elastic.clients.elasticsearch.ElasticsearchClient
4
import co.elastic.clients.elasticsearch._types.ElasticsearchException
5
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest
6
import co.elastic.clients.elasticsearch.indices.PutIndicesSettingsRequest
7
import co.elastic.clients.elasticsearch.synonyms.PutSynonymRequest
8
import java.io.StringReader
9

10
/** Index and alias management for the user-profile index. Used by applications and jobs, never by search requests. */
11
class IndexAdmin(private val client: ElasticsearchClient) {
12

13
    /** Indices an alias currently points to; empty if the alias does not exist. */
14
    fun aliasTargets(alias: String): Set<String> =
15
        if (client.indices().existsAlias { it.name(alias) }.value()) {
16
            client.indices().getAlias { it.name(alias) }.aliases().keys
17
        } else {
18
            emptySet()
19
        }
20

21
    /**
22
     * Creates `user-profile-v<version>` from its reviewed definition, aliases included.
23
     * Returns false, and changes nothing, if the index already exists.
24
     */
25
    fun createIndex(version: Int): Boolean {
26
        val name = UserProfileIndex.indexName(version)
27
        if (client.indices().exists { it.index(name) }.value()) return false
28
        ensureSynonymsSet()
29
        val request = CreateIndexRequest.of {
30
            it.index(name).withJson(StringReader(UserProfileIndex.definition(version)))
31
        }
32
        client.indices().create(request)
33
        return true
34
    }
35

36
    /** The name analysers reference the synonyms set, so it must exist before any index version. */
37
    fun ensureSynonymsSet() {
38
        val exists = try {
39
            client.synonyms().getSynonym { it.id(UserProfileIndex.SYNONYMS_SET) }
40
            true
41
        } catch (e: ElasticsearchException) {
42
            if (e.status() != 404) throw e
43
            false
44
        }
45
        if (!exists) {
46
            client.synonyms().putSynonym(
47
                PutSynonymRequest.of {
48
                    it.id(UserProfileIndex.SYNONYMS_SET).withJson(StringReader(UserProfileIndex.synonymsDefinition()))
49
                },
50
            )
51
        }
52
    }
53

54
    /** Applies dynamic index settings given as JSON, for example `{"index": {"refresh_interval": "-1"}}`. */
55
    fun putSettings(index: String, settingsJson: String) {
56
        client.indices().putSettings(
57
            PutIndicesSettingsRequest.of { it.index(index).withJson(StringReader(settingsJson)) },
58
        )
59
    }
60

61
    /** The number of replicas an index is configured with. */
62
    fun replicaCount(index: String): String =
63
        client.indices().getSettings { it.index(index) }[index]?.settings()?.index()?.numberOfReplicas()
64
            ?: error("No settings for $index")
65

66
    fun count(index: String): Long = client.count { it.index(index) }.count()
67

68
    fun refresh(index: String) {
69
        client.indices().refresh { it.index(index) }
70
    }
71

72
    /**
73
     * Moves both aliases from [fromIndex] to [toIndex] in one atomic request.
74
     * Searches and writes switch together; there is no moment when an alias points nowhere.
75
     */
76
    fun swapAliases(fromIndex: String, toIndex: String) {
77
        client.indices().updateAliases { u ->
78
            u.actions { a -> a.remove { r -> r.index(fromIndex).alias(UserProfileIndex.READ_ALIAS) } }
79
                .actions { a -> a.remove { r -> r.index(fromIndex).alias(UserProfileIndex.WRITE_ALIAS) } }
80
                .actions { a -> a.add { ad -> ad.index(toIndex).alias(UserProfileIndex.READ_ALIAS) } }
81
                .actions { a -> a.add { ad -> ad.index(toIndex).alias(UserProfileIndex.WRITE_ALIAS).isWriteIndex(true) } }
82
        }
83
    }
84
}

swapAliases sends four actions in one update aliases request: remove both aliases from the old index and add both to the new one. Elasticsearch applies the actions atomically, so there is no moment when user-profile-read points at nothing, or at both indices.

The application class, its typed configuration, and the client configuration from chapter 10:

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

3
import org.springframework.boot.autoconfigure.SpringBootApplication
4
import org.springframework.boot.context.properties.ConfigurationPropertiesScan
5
import org.springframework.boot.runApplication
6
import org.springframework.scheduling.annotation.EnableScheduling
7

8
@SpringBootApplication
9
@ConfigurationPropertiesScan
10
@EnableScheduling
11
class IndexerApplication
12

13
fun main(args: Array<String>) {
14
    runApplication<IndexerApplication>(*args)
15
}

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

3
import org.springframework.boot.context.properties.ConfigurationProperties
4
import java.time.Duration
5

6
@ConfigurationProperties("indexer")
7
data class IndexerProperties(
8
    val command: String,
9
    val sourceIndex: String,
10
    val targetIndex: String,
11
    val relay: Relay,
12
    val backfill: Backfill,
13
) {
14
    data class Relay(val pollInterval: Duration, val batchSize: Int, val targets: List<String>)
15

16
    data class Backfill(val pageSize: Int, val bulkOperations: Int, val concurrentRequests: Int)
17
}

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

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

10
@Configuration
11
class ElasticsearchConfig {
12
    @Bean
13
    fun jsonpMapper(): JsonpMapper = ElasticsearchJson.jsonpMapper()
14

15
    @Bean
16
    fun indexAdmin(client: ElasticsearchClient) = IndexAdmin(client)
17
}

Stage 2 — Reading PostgreSQL with keyset pagination

What it is. Keyset pagination reads the next page as “rows with a key greater than the last one I saw”, WHERE user_id > :after ORDER BY user_id LIMIT :limit, instead of OFFSET.

Why it matters at 300 million profiles. OFFSET 250000000 makes PostgreSQL walk and discard 250 million rows before returning the page, so every page is slower than the last and the backfill slows to a crawl near the end. A keyset page is a range scan on the primary key, and it costs the same whether it is the first page or the last.

Example. ProfileReader has three queries: a keyset page for the backfill, a lookup by IDs for the relay and for retries, and a count for validation.

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

3
import `in`.o612.eng.usersearch.index.UserProfileDocument
4
import org.springframework.jdbc.core.simple.JdbcClient
5
import org.springframework.stereotype.Component
6
import java.sql.ResultSet
7
import java.time.LocalDate
8
import java.time.OffsetDateTime
9

10
/** A profile row, with the version the index uses as its external version. */
11
data class ProfileRow(val document: UserProfileDocument, val rowVersion: Long) {
12
    val userId: Long get() = document.userId.toLong()
13
}
14

15
/** Reads profiles from PostgreSQL, the source of truth. */
16
@Component
17
class ProfileReader(private val jdbc: JdbcClient) {
18

19
    /**
20
     * Keyset pagination: the next [limit] profiles after [afterUserId], by primary key.
21
     * Every page is an index range scan, however deep the backfill is. OFFSET would re-read every skipped row.
22
     */
23
    fun page(afterUserId: Long, limit: Int): List<ProfileRow> =
24
        jdbc.sql("$SELECT WHERE user_id > :after ORDER BY user_id LIMIT :limit")
25
            .param("after", afterUserId)
26
            .param("limit", limit)
27
            .query { rs, _ -> rs.toProfileRow() }
28
            .list()
29

30
    /** The current state of specific profiles. Missing ids have been deleted. */
31
    fun byIds(userIds: Collection<Long>): Map<Long, ProfileRow> =
32
        if (userIds.isEmpty()) {
33
            emptyMap()
34
        } else {
35
            jdbc.sql("$SELECT WHERE user_id = ANY(:ids)")
36
                .param("ids", userIds.toTypedArray())
37
                .query { rs, _ -> rs.toProfileRow() }
38
                .list()
39
                .associateBy { it.userId }
40
        }
41

42
    fun count(): Long = jdbc.sql("SELECT count(*) FROM user_profile").query(Long::class.java).single()
43

44
    private fun ResultSet.toProfileRow() = ProfileRow(
45
        document = UserProfileDocument(
46
            userId = getLong("user_id").toString(),
47
            fullName = getString("full_name"),
48
            firstName = getString("first_name"),
49
            lastName = getString("last_name"),
50
            email = getString("email"),
51
            mobileNumber = getString("mobile_number"),
52
            city = getString("city"),
53
            state = getString("state"),
54
            country = getString("country"),
55
            pincode = getString("pincode"),
56
            dateOfBirth = getObject("date_of_birth", LocalDate::class.java),
57
            gender = getString("gender"),
58
            accountStatus = getString("account_status"),
59
            createdAt = getObject("created_at", OffsetDateTime::class.java).toInstant(),
60
            updatedAt = getObject("updated_at", OffsetDateTime::class.java).toInstant(),
61
        ),
62
        rowVersion = getLong("row_version"),
63
    )
64

65
    private companion object {
66
        const val SELECT = """
67
            SELECT user_id, full_name, first_name, last_name, email, mobile_number, city, state, country,
68
                   pincode, date_of_birth, gender, account_status, created_at, updated_at, row_version
69
            FROM user_profile
70
        """
71
    }
72
}

ProfileRow carries row_version next to the document, because every write to Elasticsearch uses it as the external version.

Common mistake. Paging by updated_at for a backfill. Many profiles share a timestamp, so a page boundary inside a group of equal timestamps skips or repeats rows, and rows updated during the backfill move between pages. The primary key never changes and never repeats.

Stage 3 — Classifying every bulk item

Chapter 03 established the rules for bulk item results. They become one small enum used by both jobs:

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

3
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem
4

5
/** What to do with one bulk item's result (chapter 03's table). */
6
enum class BulkOutcome {
7
    /** Written. */
8
    APPLIED,
9

10
    /** 409 on an external-version write: Elasticsearch already has this version or a newer one. */
11
    ALREADY_CURRENT,
12

13
    /** 429 or 5xx: temporary. Retry later with backoff. */
14
    RETRY,
15

16
    /** Any other failure: the request or document is wrong. Retrying will not help. */
17
    DEAD_LETTER,
18
    ;
19

20
    companion object {
21
        fun of(item: BulkResponseItem): BulkOutcome = when {
22
            item.error() == null -> APPLIED
23
            item.status() == 409 -> ALREADY_CURRENT
24
            item.status() == 429 || item.status() >= 500 -> RETRY
25
            else -> DEAD_LETTER
26
        }
27
    }
28
}

Permanent failures go to the dead-letter table with the target and the error, so they can be investigated and replayed:

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

3
import org.springframework.jdbc.core.simple.JdbcClient
4
import org.springframework.stereotype.Component
5

6
/** Changes that failed permanently. Replaying one is inserting a new outbox row for its user_id. */
7
@Component
8
class DeadLetters(private val jdbc: JdbcClient) {
9

10
    fun record(userId: Long, rowVersion: Long, target: String, errorType: String, errorReason: String?) {
11
        jdbc.sql(
12
            """
13
            INSERT INTO profile_search_dead_letter (user_id, row_version, target, error_type, error_reason)
14
            VALUES (:userId, :rowVersion, :target, :errorType, :errorReason)
15
            """,
16
        )
17
            .param("userId", userId)
18
            .param("rowVersion", rowVersion)
19
            .param("target", target)
20
            .param("errorType", errorType)
21
            .param("errorReason", errorReason)
22
            .update()
23
    }
24
}

Stage 4 — The backfill job

What it is. The backfill writes every profile into one named index, never an alias. It uses the Java client’s BulkIngester, which collects operations into bulk requests and sends them when a request reaches a number of operations or a size in bytes, 1,000 operations and 5 MiB by default.

Why it matters at 300 million profiles. A backfill is the largest write load the cluster sees. It must go as fast as the cluster allows, and no faster: an indexer that sends requests faster than the cluster can process them fills the write thread pool’s queue and gets 429 rejections. BulkIngester provides the brake. The client documentation states that when the maximum number of concurrent requests is in flight and the next request is full, adding an operation blocks. The reading loop therefore slows to the cluster’s pace. That is backpressure.

Example.

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

3
import co.elastic.clients.elasticsearch.ElasticsearchClient
4
import co.elastic.clients.elasticsearch._helpers.bulk.BulkIngester
5
import co.elastic.clients.elasticsearch._helpers.bulk.BulkListener
6
import co.elastic.clients.elasticsearch._types.VersionType
7
import co.elastic.clients.elasticsearch.core.BulkRequest
8
import co.elastic.clients.elasticsearch.core.BulkResponse
9
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation
10
import `in`.o612.eng.usersearch.index.IndexAdmin
11
import org.slf4j.LoggerFactory
12
import org.springframework.stereotype.Component
13
import java.util.concurrent.ConcurrentLinkedQueue
14
import java.util.concurrent.atomic.AtomicLong
15

16
data class BackfillReport(val read: Long, val applied: Long, val alreadyCurrent: Long, val deadLettered: Long, val retried: Long)
17

18
/**
19
 * Copies every profile from PostgreSQL into one index, by name. Never targets an alias:
20
 * a backfill fills a new index version before anything reads it.
21
 */
22
@Component
23
class BackfillJob(
24
    private val client: ElasticsearchClient,
25
    private val reader: ProfileReader,
26
    private val indexAdmin: IndexAdmin,
27
    private val deadLetters: DeadLetters,
28
    private val properties: IndexerProperties,
29
) {
30
    private val log = LoggerFactory.getLogger(javaClass)
31

32
    fun run(targetIndex: String): BackfillReport {
33
        val settings = properties.backfill
34
        val replicas = indexAdmin.replicaCount(targetIndex)
35
        // No refresh and no replicas while loading; tombstones kept long enough to outlive the load (chapter 04).
36
        indexAdmin.putSettings(targetIndex, """{"index":{"refresh_interval":"-1","number_of_replicas":0,"gc_deletes":"1h"}}""")
37

38
        val applied = AtomicLong()
39
        val alreadyCurrent = AtomicLong()
40
        val deadLettered = AtomicLong()
41
        val retry = ConcurrentLinkedQueue<Long>()
42
        var read = 0L
43

44
        val listener = object : BulkListener<ProfileRow> {
45
            override fun beforeBulk(executionId: Long, request: BulkRequest, contexts: List<ProfileRow>) = Unit
46

47
            override fun afterBulk(executionId: Long, request: BulkRequest, contexts: List<ProfileRow>, response: BulkResponse) {
48
                response.items().zip(contexts).forEach { (item, row) ->
49
                    when (BulkOutcome.of(item)) {
50
                        BulkOutcome.APPLIED -> applied.incrementAndGet()
51
                        BulkOutcome.ALREADY_CURRENT -> alreadyCurrent.incrementAndGet()
52
                        BulkOutcome.RETRY -> retry.add(row.userId)
53
                        BulkOutcome.DEAD_LETTER -> {
54
                            deadLetters.record(row.userId, row.rowVersion, targetIndex, item.error()?.type() ?: "unknown", item.error()?.reason())
55
                            deadLettered.incrementAndGet()
56
                        }
57
                    }
58
                }
59
            }
60

61
            override fun afterBulk(executionId: Long, request: BulkRequest, contexts: List<ProfileRow>, failure: Throwable) {
62
                log.warn("Bulk request {} failed as a whole; {} items will be retried: {}", executionId, contexts.size, failure.message)
63
                contexts.forEach { retry.add(it.userId) }
64
            }
65
        }
66

67
        try {
68
            // add() blocks while the maximum number of requests is in flight: that is the backpressure.
69
            BulkIngester.of<ProfileRow> {
70
                it.client(client)
71
                    .maxOperations(settings.bulkOperations)
72
                    .maxConcurrentRequests(settings.concurrentRequests)
73
                    .listener(listener)
74
            }.use { ingester ->
75
                var after = 0L
76
                while (true) {
77
                    val page = reader.page(after, settings.pageSize)
78
                    if (page.isEmpty()) break
79
                    page.forEach { row -> ingester.add(indexOperation(targetIndex, row), row) }
80
                    read += page.size
81
                    after = page.last().userId
82
                    if (read % 100_000 < settings.pageSize) log.info("Backfill read {} profiles", read)
83
                }
84
            } // close() flushes the last batch and waits for every request
85
            val retried = retryFailed(targetIndex, retry.toSet(), applied, alreadyCurrent)
86
            return BackfillReport(read, applied.get(), alreadyCurrent.get(), deadLettered.get(), retried)
87
        } finally {
88
            // Always restore, even after a failure: an index left at refresh -1 never shows new writes.
89
            // gc_deletes goes back to its documented default, 60s; the typed settings API cannot send null.
90
            indexAdmin.putSettings(targetIndex, """{"index":{"refresh_interval":"1s","number_of_replicas":$replicas,"gc_deletes":"60s"}}""")
91
            indexAdmin.refresh(targetIndex)
92
        }
93
    }
94

95
    /** Re-reads each retryable profile, so a retry writes its current state, and tries again with backoff. */
96
    private fun retryFailed(targetIndex: String, userIds: Set<Long>, applied: AtomicLong, alreadyCurrent: AtomicLong): Long {
97
        var pending = userIds
98
        var delayMillis = 1_000L
99
        repeat(MAX_RETRIES) {
100
            if (pending.isEmpty()) return userIds.size.toLong()
101
            Thread.sleep(delayMillis)
102
            delayMillis *= 2
103
            val rows = reader.byIds(pending) // deleted since: the outbox relay handles them
104
            if (rows.isEmpty()) return userIds.size.toLong()
105
            val response = client.bulk { b -> b.operations(rows.values.map { indexOperation(targetIndex, it) }) }
106
            pending = response.items().filter { item ->
107
                when (BulkOutcome.of(item)) {
108
                    BulkOutcome.APPLIED -> { applied.incrementAndGet(); false }
109
                    BulkOutcome.ALREADY_CURRENT -> { alreadyCurrent.incrementAndGet(); false }
110
                    else -> true
111
                }
112
            }.map { it.id()!!.toLong() }.toSet()
113
        }
114
        check(pending.isEmpty()) { "${pending.size} profiles still failing after $MAX_RETRIES retries" }
115
        return userIds.size.toLong()
116
    }
117

118
    private fun indexOperation(targetIndex: String, row: ProfileRow): BulkOperation = BulkOperation.of { op ->
119
        op.index { i ->
120
            i.index(targetIndex)
121
                .id(row.document.userId)
122
                .document(row.document)
123
                .version(row.rowVersion)
124
                .versionType(VersionType.External)
125
        }
126
    }
127

128
    private companion object {
129
        const val MAX_RETRIES = 3
130
    }
131
}

The job applies every rule from earlier chapters:

Settings for the load, and a guaranteed restore. Refresh is off and replicas are zero during the load, as chapter 08 recommends, and gc_deletes is raised to one hour, which closes the resurrection gap chapter 04 left open: a delete that happens during the backfill keeps its tombstone longer than the backfill’s read-to-write delay. The finally block restores all three, even when the job fails.
Contexts connect results to rows. Each operation is added with its ProfileRow as context, and the listener zips the response items with the contexts to classify each one.
Retries re-read. A 429 or 5xx item is not re-sent as it was. After the load, retryable profiles are read again from PostgreSQL, so a retry writes their current state, with exponential backoff. A profile deleted in the meantime is skipped; the relay deletes it.
External versions make it safe to repeat. Run the backfill twice and the second run applies nothing: every item returns 409, counted as already current.

Production note — The first version of this job restored gc_deletes by sending null, the usual way to reset a setting to its default. The typed settings API in client 9.4.5 rejected it while parsing the JSON (Unexpected JSON event 'VALUE_NULL'), after the load had finished, and the index was left with refresh disabled. The job now restores the documented default, 60s, explicitly. Whatever your restore mechanism, test the failure path: an index left at refresh_interval: -1 accepts writes and never shows them.

Trade-off. This backfill rebuilds from PostgreSQL. The _reindex API can copy v1 into v2 inside the cluster, which is faster and does not load PostgreSQL. It also copies whatever drift v1 has accumulated, and it cannot apply changes to how documents are built from rows. Rebuilding from the source of truth doubles as a repair; _reindex is the better choice when only the mapping changes and v1 is trusted.

Stage 5 — The outbox relay

The relay implements chapter 04’s design exactly: claim a batch with SKIP LOCKED, collapse it to one change per profile, re-read the current rows, and write each change with its row version as the external version. It commits the claim only when every item is applied or dead-lettered.

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

3
import co.elastic.clients.elasticsearch.ElasticsearchClient
4
import co.elastic.clients.elasticsearch._types.VersionType
5
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation
6
import org.slf4j.LoggerFactory
7
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
8
import org.springframework.jdbc.core.simple.JdbcClient
9
import org.springframework.scheduling.annotation.Scheduled
10
import org.springframework.stereotype.Component
11
import org.springframework.transaction.support.TransactionTemplate
12

13
data class OutboxEvent(val id: Long, val userId: Long, val rowVersion: Long, val operation: String)
14

15
/**
16
 * Drains profile_search_outbox (chapter 04). Events are notifications: the relay re-reads each profile's
17
 * current row and writes that, with its row version as the external version, to every configured target.
18
 */
19
@Component
20
@ConditionalOnProperty("indexer.command", havingValue = "relay", matchIfMissing = true)
21
class OutboxRelay(
22
    private val jdbc: JdbcClient,
23
    private val transactions: TransactionTemplate,
24
    private val reader: ProfileReader,
25
    private val client: ElasticsearchClient,
26
    private val deadLetters: DeadLetters,
27
    private val properties: IndexerProperties,
28
) {
29
    private val log = LoggerFactory.getLogger(javaClass)
30

31
    @Scheduled(fixedDelayString = "\${indexer.relay.poll-interval}")
32
    fun poll() {
33
        do {
34
            val processed = transactions.execute { relayOneBatch() } ?: 0
35
        } while (processed == properties.relay.batchSize)
36
    }
37

38
    /** Claims a batch, writes it to every target, and commits only when every item is applied or dead-lettered. */
39
    fun relayOneBatch(): Int {
40
        val events = claimBatch(properties.relay.batchSize)
41
        if (events.isEmpty()) return 0
42

43
        val rows = reader.byIds(events.map { it.userId }.toSet())
44
        val changes = events.groupBy { it.userId }.mapNotNull { (userId, userEvents) ->
45
            val row = rows[userId]
46
            val delete = userEvents.filter { it.operation == "DELETE" }.maxByOrNull { it.rowVersion }
47
            when {
48
                row != null -> Change(userId, row.rowVersion, row)
49
                delete != null -> Change(userId, delete.rowVersion, null)
50
                else -> null // deleted after these events; its DELETE event is still in the queue
51
            }
52
        }
53

54
        for (target in properties.relay.targets) {
55
            write(target, changes)
56
        }
57
        log.info("Relayed {} events as {} changes to {}", events.size, changes.size, properties.relay.targets)
58
        return events.size
59
    }
60

61
    private fun write(target: String, changes: List<Change>) {
62
        var pending = changes
63
        var delayMillis = 500L
64
        repeat(MAX_ATTEMPTS) { attempt ->
65
            if (pending.isEmpty()) return
66
            if (attempt > 0) Thread.sleep(delayMillis).also { delayMillis *= 2 }
67
            val response = client.bulk { b -> b.operations(pending.map { it.operation(target) }) }
68
            pending = response.items().zip(pending).mapNotNull { (item, change) ->
69
                when (BulkOutcome.of(item)) {
70
                    BulkOutcome.APPLIED, BulkOutcome.ALREADY_CURRENT -> null
71
                    BulkOutcome.RETRY -> change
72
                    BulkOutcome.DEAD_LETTER -> {
73
                        deadLetters.record(change.userId, change.version, target, item.error()?.type() ?: "unknown", item.error()?.reason())
74
                        null
75
                    }
76
                }
77
            }
78
        }
79
        // Still failing: roll back the transaction so the events return to the queue.
80
        check(pending.isEmpty()) { "${pending.size} changes to $target still failing after $MAX_ATTEMPTS attempts" }
81
    }
82

83
    private fun claimBatch(limit: Int): List<OutboxEvent> =
84
        jdbc.sql(
85
            """
86
            WITH batch AS (
87
                SELECT id FROM profile_search_outbox ORDER BY id LIMIT :limit FOR UPDATE SKIP LOCKED
88
            )
89
            DELETE FROM profile_search_outbox AS o
90
            USING batch
91
            WHERE o.id = batch.id
92
            RETURNING o.id, o.user_id, o.row_version, o.operation
93
            """,
94
        )
95
            .param("limit", limit)
96
            .query { rs, _ -> OutboxEvent(rs.getLong("id"), rs.getLong("user_id"), rs.getLong("row_version"), rs.getString("operation")) }
97
            .list()
98

99
    /** One profile's net change: index its current row, or delete it. */
100
    private data class Change(val userId: Long, val version: Long, val row: ProfileRow?) {
101
        fun operation(target: String): BulkOperation = BulkOperation.of { op ->
102
            if (row != null) {
103
                op.index { i ->
104
                    i.index(target).id(userId.toString()).document(row.document)
105
                        .version(version).versionType(VersionType.External)
106
                }
107
            } else {
108
                op.delete { d -> d.index(target).id(userId.toString()).version(version).versionType(VersionType.External) }
109
            }
110
        }
111
    }
112

113
    private companion object {
114
        const val MAX_ATTEMPTS = 4
115
    }
116
}

indexer.relay.targets is the list of indices or aliases every change goes to. Normally it is just user-profile-write. During a reindex it also names the new version, and after the swap, the old one. Stage 7 uses both.

Stage 6 — Commands for the reindex

Validation compares the new index with PostgreSQL and with the index it replaces:

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

3
import co.elastic.clients.elasticsearch.ElasticsearchClient
4
import co.elastic.clients.elasticsearch._types.FieldValue
5
import co.elastic.clients.elasticsearch._types.SortOrder
6
import co.elastic.clients.elasticsearch._types.query_dsl.Operator
7
import `in`.o612.eng.usersearch.index.IndexAdmin
8
import org.springframework.stereotype.Component
9

10
/** Checks a new index version against PostgreSQL and against the version it replaces. */
11
@Component
12
class ReindexValidator(
13
    private val client: ElasticsearchClient,
14
    private val indexAdmin: IndexAdmin,
15
    private val reader: ProfileReader,
16
) {
17
    /** Sample searches that must return the same profiles, in the same order, from both versions. */
18
    private val samples = listOf("prashant kumar", "mohd", "jose fernandes", "priya", "lakshmi iyer")
19

20
    fun validate(sourceIndex: String, targetIndex: String): Boolean {
21
        indexAdmin.refresh(targetIndex)
22
        val expected = reader.count()
23
        val actual = indexAdmin.count(targetIndex)
24
        println("count  postgres=$expected  $targetIndex=$actual  ${if (expected == actual) "OK" else "MISMATCH"}")
25
        var ok = expected == actual
26

27
        for (name in samples) {
28
            val before = topIds(sourceIndex, name)
29
            val after = topIds(targetIndex, name)
30
            val same = before == after
31
            println("search '$name'  total ${before.first}/${after.first}  top-10 ${if (same) "identical" else "DIFFERENT"}")
32
            ok = ok && same
33
        }
34
        return ok
35
    }
36

37
    private fun topIds(index: String, name: String): Pair<Long, List<String>> {
38
        val response = client.search({ s ->
39
            s.index(index).size(10).trackTotalHits { it.enabled(true) }.source { it.fetch(false) }
40
                .query { q ->
41
                    q.bool { b ->
42
                        b.must { m -> m.match { mt -> mt.field("fullName").query(name).operator(Operator.And) } }
43
                            .filter { f -> f.term { t -> t.field("accountStatus").value(FieldValue.of("ACTIVE")) } }
44
                    }
45
                }
46
                .sort { it.score { sc -> sc.order(SortOrder.Desc) } }
47
                .sort { it.field { f -> f.field("updatedAt").order(SortOrder.Desc) } }
48
                .sort { it.field { f -> f.field("userId").order(SortOrder.Asc) } }
49
        }, Void::class.java)
50
        return (response.hits().total()?.value() ?: 0) to response.hits().hits().map { it.id()!! }
51
    }
52
}

The sample searches must return the same totals and the same top ten, in the same order, from both versions. That checks that nothing in the mapping change altered search results for the shapes that matter. Extend the list with your own recorded queries.

The command runner dispatches create, backfill, validate, and swap, and exits with a non-zero status when a command fails, so a deployment pipeline can stop on it:

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

3
import `in`.o612.eng.usersearch.index.IndexAdmin
4
import `in`.o612.eng.usersearch.index.UserProfileIndex
5
import org.springframework.boot.ApplicationArguments
6
import org.springframework.boot.ApplicationRunner
7
import org.springframework.boot.SpringApplication
8
import org.springframework.context.ApplicationContext
9
import org.springframework.stereotype.Component
10
import kotlin.system.exitProcess
11

12
/** Runs a one-off command (create, backfill, validate, swap) and exits. The default, relay, keeps running. */
13
@Component
14
class CommandRunner(
15
    private val properties: IndexerProperties,
16
    private val backfill: BackfillJob,
17
    private val validator: ReindexValidator,
18
    private val indexAdmin: IndexAdmin,
19
    private val context: ApplicationContext,
20
) : ApplicationRunner {
21

22
    override fun run(args: ApplicationArguments) {
23
        val succeeded = when (properties.command) {
24
            "relay" -> return
25
            "create" -> {
26
                val version = requireTarget().substringAfterLast("-v").toInt()
27
                println("created ${requireTarget()}: ${indexAdmin.createIndex(version)}")
28
                true
29
            }
30
            "backfill" -> {
31
                val report = backfill.run(requireTarget())
32
                println("backfill $report")
33
                report.deadLettered == 0L
34
            }
35
            "validate" -> validator.validate(requireSource(), requireTarget())
36
            "swap" -> {
37
                indexAdmin.swapAliases(requireSource(), requireTarget())
38
                println("read -> ${indexAdmin.aliasTargets(UserProfileIndex.READ_ALIAS)}, write -> ${indexAdmin.aliasTargets(UserProfileIndex.WRITE_ALIAS)}")
39
                true
40
            }
41
            else -> error("Unknown indexer.command '${properties.command}'")
42
        }
43
        exitProcess(SpringApplication.exit(context, { if (succeeded) 0 else 1 }))
44
    }
45

46
    private fun requireSource() = properties.sourceIndex.ifBlank { error("Set indexer.source-index") }
47

48
    private fun requireTarget() = properties.targetIndex.ifBlank { error("Set indexer.target-index") }
49
}

Stage 7 — Start the relay

Every command below runs from the user-search directory. Export the connection settings once; the database password comes from the lab’s .env:

1
set -a; source ../user-search-lab/.env; set +a
2
export DATABASE_URL=jdbc:postgresql://localhost:5432/profiles
3
export DATABASE_USERNAME=profiles DATABASE_PASSWORD="$POSTGRES_PASSWORD"
4
export ELASTICSEARCH_URIS=https://localhost:9200
5
export ELASTICSEARCH_USERNAME=elastic ELASTICSEARCH_PASSWORD="$ELASTIC_PASSWORD"
6
export ELASTICSEARCH_CA_CERT="file:$(cd ../user-search-lab && pwd)/certs/ca/ca.crt"
7
./gradlew :indexer:bootRun

The outbox has held seven events since chapter 04. Within a second:

1
INFO ... c.e.usersearch.indexer.OutboxRelay : Relayed 7 events as 5 changes to [user-profile-write]

Seven events became five changes because profile 42’s three updates collapsed into one. Now change PostgreSQL in another terminal:

1
UPDATE user_profile SET account_status = 'INACTIVE', updated_at = now() WHERE user_id = 1;
2
DELETE FROM user_profile WHERE user_id = 2;

1
INFO ... c.e.usersearch.indexer.OutboxRelay : Relayed 2 events as 2 changes to [user-profile-write]

GET user-profile-read/_doc/1 returns "_version":2 and "accountStatus":"INACTIVE", matching the row’s row_version of 2; GET user-profile-read/_doc/2 returns "found":false. The search projection now follows PostgreSQL within about a second, with no application code writing to Elasticsearch.

Stage 8 — Reindex from v1 to v2 without downtime

The procedure has six steps. Searches and writes continue throughout.

The diagram shows the six steps in order: create v2, backfill it while the relay writes every change to both versions, wait until the outbox is drained, validate, swap both aliases in one request, and keep v1 current for a rollback window before deleting it. A failed validation loops back to fixing the cause and running the backfill again.

a. Create v2. From its reviewed definition:

1
./gradlew :indexer:bootRun --args='--indexer.command=create --indexer.target-index=user-profile-v2'

1
created user-profile-v2: true

b. Backfill v2 while the relay writes to both. Stop the relay and restart it with v2 as a second target, so every change from now on reaches both versions:

1
INDEXER_RELAY_TARGETS=user-profile-write,user-profile-v2 ./gradlew :indexer:bootRun

In a second terminal, start the backfill:

1
./gradlew :indexer:bootRun --args='--indexer.command=backfill --indexer.target-index=user-profile-v2'

While it runs, change three profiles in PostgreSQL: one that the backfill has already passed, one it has not reached, and one deletion:

1
UPDATE user_profile SET account_status = 'SUSPENDED', updated_at = now() WHERE user_id = 3;
2
DELETE FROM user_profile WHERE user_id = 5;
3
UPDATE user_profile SET city = 'Kochi', updated_at = now() WHERE user_id = 999999;

The relay applies them to both indices while the backfill continues. The backfill ends with a report. This one comes from a run into an empty copy of v2 at the end of the lab session, when 999,997 profiles remained; your read count reflects the profiles you have created and deleted:

1
backfill BackfillReport(read=999997, applied=999997, alreadyCurrent=0, deadLettered=0, retried=0)

In the lab, reading and indexing about a million profiles took about 96 seconds, including application startup. Your number will differ. A second run of the same command reports applied=0 and alreadyCurrent equal to every profile.

c. Catch up. Changes made during the backfill have already reached v2 through the relay. Before swapping, confirm the queue is empty:

1
SELECT count(*) FROM profile_search_outbox;

1
 count
2
-------
3
     0

The three changed profiles agree in PostgreSQL and in both indices: profile 3 is SUSPENDED at version 2, profile 5 is absent, and profile 999999 is in Kochi at version 2. The backfill had read profile 999999 after its update, and profile 3 before it; external versions made the order irrelevant.

d. Validate.

1
./gradlew :indexer:bootRun --args='--indexer.command=validate --indexer.source-index=user-profile-v1 --indexer.target-index=user-profile-v2'

1
count  postgres=999998  user-profile-v2=999998  OK
2
search 'prashant kumar'  total 1191/1191  top-10 identical
3
search 'mohd'  total 23333/23333  top-10 identical
4
search 'jose fernandes'  total 1170/1170  top-10 identical
5
search 'priya'  total 23384/23384  top-10 identical
6
search 'lakshmi iyer'  total 1164/1164  top-10 identical

Your counts differ by the profiles you created and deleted; what matters is that PostgreSQL and v2 agree, and every sample search is identical.

e. Swap. One request moves both aliases:

1
./gradlew :indexer:bootRun --args='--indexer.command=swap --indexer.source-index=user-profile-v1 --indexer.target-index=user-profile-v2'

1
read -> [user-profile-v2], write -> [user-profile-v2]

The Search API now reads v2. It was not restarted and did not notice.

f. Keep v1 current for rollback. Restart the relay with v1 as the extra target, so the old version keeps receiving every change, including deletions, for as long as a rollback might be needed:

1
INDEXER_RELAY_TARGETS=user-profile-write,user-profile-v1 ./gradlew :indexer:bootRun

A deletion made now reaches both indices. A rollback is the same swap command with source and target exchanged, and it is instant. When the rollback window closes, stop writing to v1 and delete it with DELETE user-profile-v1.

Security note — An old index version is a copy of personal data. If you keep v1 without updating it, a profile erased from PostgreSQL survives in v1, and a rollback would bring it back into search results. Either keep writing to the retained version, as step f does, or delete it promptly. Chapter 16 covers erasure across index versions and snapshots.

Stage 9 — Use the new fields

The Search API’s facets can now count the display sub-fields. In UserFacetsBuilder, add the field mapping after BUCKETS_PER_FACET, and use it in the terms aggregation:

1
    /** Facets count the display sub-fields, which keep the original case (user-profile-v2, chapter 14). */
2
    private val AGGREGATION_FIELDS = mapOf("state" to "state.display", "city" to "city.display", "gender" to "gender")

1
                .aggregations("values", Aggregation.of { t -> t.terms { tt -> tt.field(AGGREGATION_FIELDS.getValue(field)).size(BUCKETS_PER_FACET) } })

Restart search-api:

1
curl -s 'localhost:8080/api/users/facets?state=bihar&city=patna'

1
{"facets":{"state":[{"value":"Bihar","count":70214}],"city":[{"value":"Patna","count":70214},{"value":"Gaya","count":69711}],
2
 "gender":[{"value":"FEMALE","count":34752},{"value":"MALE","count":34707},{"value":"OTHER","count":755}]}}

The counts are the same as in chapter 12; the values are now the ones stored in PostgreSQL.

Production note — Deploy this change only after the swap. Against v1, which has no display sub-fields, the facets would return empty lists, not an error. Code that depends on a new mapping ships after the index that provides it, and a rollback to v1 needs the previous code too.

The new index is also smaller. GET _cat/indices/user-profile-*?v&h=index,docs.count,store.size reported 150.5 MB for v2, against the 157.6 MB chapter 05 measured for v1 when freshly loaded: the doc values removed from email and mobileNumber outweigh the new display sub-fields.

Checkpoint

Both aliases point to user-profile-v2, the outbox is empty, validate reports OK and every sample identical, and the facets endpoint returns Bihar rather than bihar. If the swap command fails, nothing changed: the aliases still point to v1, because the four actions are applied together or not at all.

Common mistakes with ingestion and reindexing

OFFSET pagination for the backfill. It slows down with every page. Use the primary key.
Ignoring bulk item results. An HTTP 200 can hide rejected documents.
Retrying rejected items as they were. Re-read them, so the retry writes their current state.
Leaving load settings in place. Restore refresh, replicas, and gc_deletes in a finally block, and test that it runs.
Backfilling through an alias. Write the new version by name, so nothing reads it until validation has passed.
Swapping without catching up. Changes made during the backfill must reach the new index before it serves searches.
Keeping the old version stale. A rollback to a stale index serves old and deleted profiles.

What you built, and what comes next

The indexer application owns every write to Elasticsearch. Its relay keeps the projection within a second or so of PostgreSQL, with re-reads, external versions, retries, and dead letters. Its backfill rebuilds any index version from the source of truth with backpressure and a guaranteed settings restore. With them you moved the live system from v1 to v2, with concurrent changes, validation, and an atomic swap, while the Search API kept serving.

The relay runs as a single process here. Several instances can share the outbox safely because of SKIP LOCKED and external versions, as chapter 04 argued, but this chapter did not run them concurrently.

Chapter 15 measures what the settings in this chapter and earlier ones actually buy: refresh and replicas during indexing, filter caching and request caching during search, index sorting, and the profile API for finding out where a slow query spends its time.

ElasticsearchSpring BootKotlinPostgres