forked from elastic/elasticsearch
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Use follower primary term when applying operations
The primary shard copy on the following has authority of the replication operations that occur on the following side in cross-cluster replication. Yet today we are using the primary term directly from the operations on the leader side. Instead we should be replacing the primary term on the following side with the primary term of the primary on the following side. This commit does this by copying the translog operations with the corrected primary term. This ensures that we use this primary term while applying the operations on the primary, and when replicating them across to the replica (where the replica request was carrying the primary term of the primary shard copy on the follower).
- Loading branch information
1 parent
530089f
commit ddae214
Showing
4 changed files
with
199 additions
and
38 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
82 changes: 82 additions & 0 deletions
82
...n/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/bulk/BulkShardOperationsTests.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.ccr.action.bulk; | ||
|
|
||
| import org.apache.lucene.index.Term; | ||
| import org.elasticsearch.action.support.replication.TransportWriteAction; | ||
| import org.elasticsearch.cluster.routing.RecoverySource; | ||
| import org.elasticsearch.cluster.routing.ShardRouting; | ||
| import org.elasticsearch.cluster.routing.ShardRoutingState; | ||
| import org.elasticsearch.cluster.routing.TestShardRouting; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.index.VersionType; | ||
| import org.elasticsearch.index.mapper.Uid; | ||
| import org.elasticsearch.index.shard.IndexShard; | ||
| import org.elasticsearch.index.shard.IndexShardTestCase; | ||
| import org.elasticsearch.index.shard.ShardId; | ||
| import org.elasticsearch.index.translog.Translog; | ||
| import org.elasticsearch.xpack.ccr.CcrSettings; | ||
| import org.elasticsearch.xpack.ccr.index.engine.FollowingEngineFactory; | ||
|
|
||
| import java.io.IOException; | ||
| import java.nio.charset.StandardCharsets; | ||
|
|
||
| import static org.hamcrest.Matchers.equalTo; | ||
|
|
||
| public class BulkShardOperationsTests extends IndexShardTestCase { | ||
|
|
||
| private static final byte[] SOURCE = "{}".getBytes(StandardCharsets.UTF_8); | ||
|
|
||
| // test that we use the primary term on the follower when applying operations from the leader | ||
| public void testPrimaryTermFromFollower() throws IOException { | ||
| final Settings settings = Settings.builder().put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true).build(); | ||
| final IndexShard followerPrimary = newStartedShard(true, settings, new FollowingEngineFactory()); | ||
|
|
||
| // we use this primary on the operations yet we expect the applied operations to have the primary term of the follower | ||
| final long primaryTerm = randomLongBetween(1, followerPrimary.getPrimaryTerm()); | ||
|
|
||
| final Translog.Operation[] operations = new Translog.Operation[randomIntBetween(0, 127)]; | ||
| for (int i = 0; i < operations.length; i++) { | ||
| final String id = Integer.toString(i); | ||
| final long seqNo = i; | ||
| final Translog.Operation.Type type = | ||
| randomValueOtherThan(Translog.Operation.Type.CREATE, () -> randomFrom(Translog.Operation.Type.values())); | ||
| switch (type) { | ||
| case INDEX: | ||
| operations[i] = new Translog.Index("_doc", id, seqNo, primaryTerm, 0, VersionType.INTERNAL, SOURCE, null, -1); | ||
| break; | ||
| case DELETE: | ||
| operations[i] = | ||
| new Translog.Delete("_doc", id, new Term("_id", Uid.encodeId(id)), seqNo, primaryTerm, 0, VersionType.INTERNAL); | ||
| break; | ||
| case NO_OP: | ||
| operations[i] = new Translog.NoOp(seqNo, primaryTerm, "test"); | ||
| break; | ||
| default: | ||
| throw new IllegalStateException("unexpected operation type [" + type + "]"); | ||
| } | ||
| } | ||
|
|
||
| final TransportWriteAction.WritePrimaryResult<BulkShardOperationsRequest, BulkShardOperationsResponse> result = | ||
| TransportBulkShardOperationsAction.shardOperationOnPrimary(followerPrimary.shardId(), operations, followerPrimary, logger); | ||
|
|
||
| try (Translog.Snapshot snapshot = followerPrimary.newTranslogSnapshotFromMinSeqNo(0)) { | ||
| assertThat(snapshot.totalOperations(), equalTo(operations.length)); | ||
| Translog.Operation operation; | ||
| while ((operation = snapshot.next()) != null) { | ||
| assertThat(operation.primaryTerm(), equalTo(followerPrimary.getPrimaryTerm())); | ||
| } | ||
| } | ||
|
|
||
| for (final Translog.Operation operation : result.replicaRequest().getOperations()) { | ||
| assertThat(operation.primaryTerm(), equalTo(followerPrimary.getPrimaryTerm())); | ||
| } | ||
|
|
||
| closeShards(followerPrimary); | ||
| } | ||
|
|
||
| } |