Elasticsearch upgrade using Kafka

Recently, we upgraded our Elasticsearch clusters from version 6 to version 7. This blog post describes how we did it.

Why upgrade

The upgrade was driven by the desire to search using Dense Vectors when searching unstructured content. After upgrading that cluster, we started to think about the most effective way to upgrade the remaining Elasticsearch clusters used for other purposes. It’s important to not get too far behind the upgrade cycle, and our analytics clusters could benefit from the new circuit breaker support , since we’ve had problems from time to time with large queries having negative impact on the cluster.

Goals for the upgrade process

We had some goals in our mind while planning the upgrade:

• There should be zero downtime/service interruption.
• If we see performance degradation or errors, we should be able to revert without data loss.

How we did it

Although Elastic has a documented upgrade process , we decided against that approach because of the potential for downtime and the difficulty of reverting in case of trouble. Instead of upgrading the cluster in place, we provisioned a new Elasticsearch 7 cluster and began indexing documents in both, a process commonly referred to as “dual-write”. There was one difference, though – instead of pushing documents directly to Elasticsearch 7, we wrote them to Kafka and loaded them asynchronously. Direct dual-writing to two different clusters proved out to be risky and error-prone in local testing. Writing to Kafka prevents problems with the new cluster from affecting the existing indexing process, and we were also interested to gain some advantages from this architecture:

• Kafka MirrorMaker is a more efficient way to transfer data to remote regions than making individual HTTP requests.
• For disaster recovery, we can replay the Kafka topic to re-generate an Elasticsearch index, which is much faster than re-generating all of the documents from source systems.

After we enabled dual-write and ran a full load, we observed the same number of documents in both the clusters. Then, we verified the contents of the query responses. The response verification process was service dependent. Some of our services were simple enough that we could run the same queries on both the clusters and compare the responses manually. For the more complicated ones, we had to employ an automatic verification process. We always use a dedicated RPC server to translate and execute queries against Elasticsearch to avoid the challenges traditionally associated with shared databases. By capturing the queries and responses, we can replay traffic from the production Elasticsearch 6 cluster against the Elasticsearch 7 cluster under test. Once it was verified that they return the same response, we routed the traffic to the Elasticsearch 7 cluster.

Since Elasticsearch REST client 6.8.12 supports both Elasticsearch 6 and 7, we could use the same code for both the clusters.

The nuts and bolts

At first, we made sure that the source and target cluster had the same set of indices and aliases using the Index management API.

The next step was to update the data. The data in an Elasticsearch cluster can be updated by the following ways:

Out of the above options, only IndexRequest and DeleteRequest are idempotent. Since we were using Kafka, the same message may be replayed multiple times during application deployment and scaling. As a result, idempotence of the messages was required. UpdateRequest, UpdateByQuery and DeleteByQuery didn’t produce the same result if replayed multiple times. In fact, their output was not even deterministic. As a result, we changed our code to only use these two type of requests. Fortunately, most of our code already conformed to that. The next step was to serialize the IndexRequest and DeleteRequest to a Kafka topic. We set up a loader daemon that read from a Kafka topic and write to an Elasticsearch 7 cluster asynchronously. Since the same set of IndexRequest and DeleteRequest were sent to both the clusters (they had the same set of indices and aliases), they ended up being eventually consistent.

Since the two clusters had the same set of documents, we could hot-swap them. As a result, we didn’t experience any downtime or service disruption during the process. For the same reason, we could also swap them back in case of any service degradation. The most interesting part was that the solution can be applied for any future upgrade as well.

What we learned

• Any cluster can be upgraded by following the above technique. We have tested it for clusters with 100s of millions of documents.
• The REST API compatibility between two major versions is the primary reason for the seamless upgrade experience. Based on the current setup, we can upgrade more frequently and without too much throw-away work.