Skip to content

TEST: write ops should execute under shard permit #28966

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Mar 15, 2018
Merged

TEST: write ops should execute under shard permit #28966

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4f92e6a
TEST: write ops should execute under shard permit
dnhatn Mar 9, 2018
0e2c4c4
Merge branch 'master' into test-write-under-shard-permit
dnhatn Mar 15, 2018
bad0379
private replication group + rename acquire
dnhatn Mar 15, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 21 additions & 8 deletions ...er/src/test/java/org/elasticsearch/index/replication/ESIndexLevelReplicationTestCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -456,6 +456,10 @@ public void onFailure(Exception e) {
}
}

IndexShard getPrimaryShard() {
return replicationGroup.primary;
}

protected abstract PrimaryResult performOnPrimary(IndexShard primary, Request request) throws Exception;

protected abstract void performOnReplica(ReplicaRequest request, IndexShard replica) throws Exception;
Expand Down Expand Up @@ -592,7 +596,7 @@ protected PrimaryResult performOnPrimary(IndexShard primary, BulkShardRequest re

@Override
protected void performOnReplica(BulkShardRequest request, IndexShard replica) throws Exception {
executeShardBulkOnReplica(replica, request);
executeShardBulkOnReplica(request, replica, getPrimaryShard().getPrimaryTerm(), getPrimaryShard().getGlobalCheckpoint());
}
}

Expand All @@ -602,15 +606,24 @@ private TransportWriteAction.WritePrimaryResult<BulkShardRequest, BulkShardRespo
((IndexRequest) itemRequest.request()).process(Version.CURRENT, null, index.getName());
}
}
final TransportWriteAction.WritePrimaryResult<BulkShardRequest, BulkShardResponse> result =
TransportShardBulkAction.performOnPrimary(request, primary, null,
System::currentTimeMillis, new TransportShardBulkActionTests.NoopMappingUpdatePerformer());
final PlainActionFuture<Releasable> permitAcquiredFuture = new PlainActionFuture<>();
primary.acquirePrimaryOperationPermit(permitAcquiredFuture, ThreadPool.Names.SAME, request);
final TransportWriteAction.WritePrimaryResult<BulkShardRequest, BulkShardResponse> result;
try (Releasable ignored = permitAcquiredFuture.actionGet()) {
result = TransportShardBulkAction.performOnPrimary(request, primary, null, System::currentTimeMillis,
new TransportShardBulkActionTests.NoopMappingUpdatePerformer());
}
TransportWriteActionTestHelper.performPostWriteActions(primary, request, result.location, logger);
return result;
}

private void executeShardBulkOnReplica(IndexShard replica, BulkShardRequest request) throws Exception {
final Translog.Location location = TransportShardBulkAction.performOnReplica(request, replica);
private void executeShardBulkOnReplica(BulkShardRequest request, IndexShard replica, long operationPrimaryTerm, long globalCheckpointOnPrimary) throws Exception {
final PlainActionFuture<Releasable> permitAcquiredFuture = new PlainActionFuture<>();
replica.acquireReplicaOperationPermit(operationPrimaryTerm, globalCheckpointOnPrimary, permitAcquiredFuture, ThreadPool.Names.SAME, request);
final Translog.Location location;
try (Releasable ignored = permitAcquiredFuture.actionGet()) {
location = TransportShardBulkAction.performOnReplica(request, replica);
}
TransportWriteActionTestHelper.performPostWriteActions(replica, request, location, logger);
}

Expand All @@ -630,8 +643,8 @@ BulkShardRequest indexOnPrimary(IndexRequest request, IndexShard primary) throws
/**
* indexes the given requests on the supplied replica shard
*/
void indexOnReplica(BulkShardRequest request, IndexShard replica) throws Exception {
executeShardBulkOnReplica(replica, request);
void indexOnReplica(BulkShardRequest request, ReplicationGroup group, IndexShard replica) throws Exception {
executeShardBulkOnReplica(request, replica, group.primary.getPrimaryTerm(), group.primary.getGlobalCheckpoint());
}

class GlobalCheckpointSync extends ReplicationAction<
Expand Down
4 changes: 2 additions & 2 deletions server/src/test/java/org/elasticsearch/index/replication/IndexLevelReplicationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,7 +209,7 @@ public void testConflictingOpsOnReplica() throws Exception {
logger.info("--> isolated replica " + replica1.routingEntry());
BulkShardRequest replicationRequest = indexOnPrimary(indexRequest, shards.getPrimary());
for (int i = 1; i < replicas.size(); i++) {
indexOnReplica(replicationRequest, replicas.get(i));
indexOnReplica(replicationRequest, shards, replicas.get(i));
}

logger.info("--> promoting replica to primary " + replica1.routingEntry());
Expand Down Expand Up @@ -318,7 +318,7 @@ public void testSeqNoCollision() throws Exception {
logger.info("--> Isolate replica1");
IndexRequest indexDoc1 = new IndexRequest(index.getName(), "type", "d1").source("{}", XContentType.JSON);
BulkShardRequest replicationRequest = indexOnPrimary(indexDoc1, shards.getPrimary());
indexOnReplica(replicationRequest, replica2);
indexOnReplica(replicationRequest, shards, replica2);

final Translog.Operation op1;
final List<Translog.Operation> initOperations = new ArrayList<>(initDocs);
Expand Down
6 changes: 3 additions & 3 deletions server/src/test/java/org/elasticsearch/index/replication/RecoveryDuringReplicationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ public void testRecoveryAfterPrimaryPromotion() throws Exception {
final IndexRequest indexRequest = new IndexRequest(index.getName(), "type", "rollback_" + i)
.source("{}", XContentType.JSON);
final BulkShardRequest bulkShardRequest = indexOnPrimary(indexRequest, oldPrimary);
indexOnReplica(bulkShardRequest, replica);
indexOnReplica(bulkShardRequest, shards, replica);
}
if (randomBoolean()) {
oldPrimary.flush(new FlushRequest(index.getName()));
Expand Down Expand Up @@ -326,7 +326,7 @@ public void testReplicaRollbackStaleDocumentsInPeerRecovery() throws Exception {
final IndexRequest indexRequest = new IndexRequest(index.getName(), "type", "stale_" + i)
.source("{}", XContentType.JSON);
final BulkShardRequest bulkShardRequest = indexOnPrimary(indexRequest, oldPrimary);
indexOnReplica(bulkShardRequest, replica);
indexOnReplica(bulkShardRequest, shards, replica);
}
shards.flush();
shards.promoteReplicaToPrimary(newPrimary).get();
Expand Down Expand Up @@ -374,7 +374,7 @@ public void testResyncAfterPrimaryPromotion() throws Exception {
final IndexRequest indexRequest = new IndexRequest(index.getName(), "type", "extra_" + i)
.source("{}", XContentType.JSON);
final BulkShardRequest bulkShardRequest = indexOnPrimary(indexRequest, oldPrimary);
indexOnReplica(bulkShardRequest, newPrimary);
indexOnReplica(bulkShardRequest, shards, newPrimary);
}
logger.info("--> resyncing replicas");
PrimaryReplicaSyncer.ResyncTask task = shards.promoteReplicaToPrimary(newPrimary).get();
Expand Down