Skip to content

Commit 50acab7

Browse files
DaveCTurnerDaveCTurner
committed
Revert API change ReplicationResponse -> Void
1 parent 2e0e555 commit 50acab7

File tree

13 files changed

+88
-86
lines changed

13 files changed

+88
-86
lines changed

‎server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import com.carrotsearch.hppc.ObjectLongMap;
2424
import org.elasticsearch.Version;
2525
import org.elasticsearch.action.ActionListener;
26+
import org.elasticsearch.action.support.replication.ReplicationResponse;
2627
import org.elasticsearch.cluster.routing.AllocationId;
2728
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
2829
import org.elasticsearch.cluster.routing.ShardRouting;
@@ -161,7 +162,7 @@ public class ReplicationTracker extends AbstractIndexShardComponent implements L
161162
* A callback when a new retention lease is created or an existing retention lease is removed. In practice, this callback invokes the
162163
* retention lease sync action, to sync retention leases to replicas.
163164
*/
164-
private final BiConsumer<RetentionLeases, ActionListener<Void>> onSyncRetentionLeases;
165+
private final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onSyncRetentionLeases;
165166

166167
/**
167168
* This set contains allocation IDs for which there is a thread actively waiting for the local checkpoint to advance to at least the
@@ -232,7 +233,7 @@ public RetentionLease addRetentionLease(
232233
final String id,
233234
final long retainingSequenceNumber,
234235
final String source,
235-
final ActionListener<Void> listener) {
236+
final ActionListener<ReplicationResponse> listener) {
236237
Objects.requireNonNull(listener);
237238
final RetentionLease retentionLease;
238239
final RetentionLeases currentRetentionLeases;
@@ -291,7 +292,7 @@ public synchronized RetentionLease renewRetentionLease(final String id, final lo
291292
* @param id the identifier of the retention lease
292293
* @param listener the callback when the retention lease is successfully removed and synced to replicas
293294
*/
294-
public void removeRetentionLease(final String id, final ActionListener<Void> listener) {
295+
public void removeRetentionLease(final String id, final ActionListener<ReplicationResponse> listener) {
295296
Objects.requireNonNull(listener);
296297
final RetentionLeases currentRetentionLeases;
297298
synchronized (this) {
@@ -632,7 +633,7 @@ public ReplicationTracker(
632633
final long globalCheckpoint,
633634
final LongConsumer onGlobalCheckpointUpdated,
634635
final LongSupplier currentTimeMillisSupplier,
635-
final BiConsumer<RetentionLeases, ActionListener<Void>> onSyncRetentionLeases) {
636+
final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onSyncRetentionLeases) {
636637
super(shardId, indexSettings);
637638
assert globalCheckpoint >= SequenceNumbers.UNASSIGNED_SEQ_NO : "illegal initial global checkpoint: " + globalCheckpoint;
638639
this.shardAllocationId = allocationId;

‎server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ public RetentionLeaseBackgroundSyncAction(
102102
public void backgroundSync(
103103
final ShardId shardId,
104104
final RetentionLeases retentionLeases,
105-
final ActionListener<Void> listener) {
105+
final ActionListener<ReplicationResponse> listener) {
106106
Objects.requireNonNull(shardId);
107107
Objects.requireNonNull(retentionLeases);
108108
final ThreadContext threadContext = threadPool.getThreadContext();
@@ -112,7 +112,7 @@ public void backgroundSync(
112112
execute(
113113
new Request(shardId, retentionLeases),
114114
ActionListener.wrap(
115-
r -> listener.onResponse(null),
115+
listener::onResponse,
116116
e -> {
117117
if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class, IndexShardClosedException.class) == null) {
118118
getLogger().warn(new ParameterizedMessage("{} retention lease background sync failed", shardId), e);

‎server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseSyncAction.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -101,7 +101,7 @@ public RetentionLeaseSyncAction(
101101
public void sync(
102102
final ShardId shardId,
103103
final RetentionLeases retentionLeases,
104-
final ActionListener<Void> listener) {
104+
final ActionListener<ReplicationResponse> listener) {
105105
Objects.requireNonNull(shardId);
106106
Objects.requireNonNull(retentionLeases);
107107
Objects.requireNonNull(listener);
@@ -112,7 +112,7 @@ public void sync(
112112
execute(
113113
new RetentionLeaseSyncAction.Request(shardId, retentionLeases),
114114
ActionListener.wrap(
115-
r -> listener.onResponse(null),
115+
listener::onResponse,
116116
e -> {
117117
if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class, IndexShardClosedException.class) == null) {
118118
getLogger().warn(new ParameterizedMessage("{} retention lease sync failed", shardId), e);

‎server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseSyncer.java‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
package org.elasticsearch.index.seqno;
2121

2222
import org.elasticsearch.action.ActionListener;
23+
import org.elasticsearch.action.support.replication.ReplicationResponse;
2324
import org.elasticsearch.index.shard.ShardId;
2425

2526
/**
@@ -36,7 +37,7 @@ public interface RetentionLeaseSyncer {
3637
* @param retentionLeases the retention leases to sync
3738
* @param listener the callback when sync completes
3839
*/
39-
void sync(ShardId shardId, RetentionLeases retentionLeases, ActionListener<Void> listener);
40+
void sync(ShardId shardId, RetentionLeases retentionLeases, ActionListener<ReplicationResponse> listener);
4041

4142
/**
4243
* Represents a method that when invoked syncs retention leases to replica shards in the background.
@@ -46,17 +47,18 @@ public interface RetentionLeaseSyncer {
4647
* @param retentionLeases the retention leases to sync
4748
* @param listener the callback when sync completes
4849
*/
49-
void backgroundSync(ShardId shardId, RetentionLeases retentionLeases, ActionListener<Void> listener);
50+
void backgroundSync(ShardId shardId, RetentionLeases retentionLeases, ActionListener<ReplicationResponse> listener);
5051

5152
RetentionLeaseSyncer EMPTY = new RetentionLeaseSyncer() {
5253
@Override
53-
public void sync(final ShardId shardId, final RetentionLeases retentionLeases, final ActionListener<Void> listener) {
54-
listener.onResponse(null);
54+
public void sync(final ShardId shardId, final RetentionLeases retentionLeases, final ActionListener<ReplicationResponse> listener) {
55+
listener.onResponse(new ReplicationResponse());
5556
}
5657

5758
@Override
58-
public void backgroundSync(final ShardId shardId, final RetentionLeases retentionLeases, ActionListener<Void> listener) {
59-
listener.onResponse(null);
59+
public void backgroundSync(final ShardId shardId, final RetentionLeases retentionLeases,
60+
final ActionListener<ReplicationResponse> listener) {
61+
listener.onResponse(new ReplicationResponse());
6062
}
6163
};
6264

‎server/src/main/java/org/elasticsearch/index/shard/IndexShard.java‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
import org.elasticsearch.action.admin.indices.flush.FlushRequest;
4242
import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
4343
import org.elasticsearch.action.admin.indices.upgrade.post.UpgradeRequest;
44+
import org.elasticsearch.action.support.replication.ReplicationResponse;
4445
import org.elasticsearch.cluster.metadata.IndexMetaData;
4546
import org.elasticsearch.cluster.metadata.MappingMetaData;
4647
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
@@ -1938,7 +1939,7 @@ public RetentionLease addRetentionLease(
19381939
final String id,
19391940
final long retainingSequenceNumber,
19401941
final String source,
1941-
final ActionListener<Void> listener) {
1942+
final ActionListener<ReplicationResponse> listener) {
19421943
Objects.requireNonNull(listener);
19431944
assert assertPrimaryMode();
19441945
verifyNotClosed();
@@ -1978,7 +1979,7 @@ public RetentionLease renewRetentionLease(final String id, final long retainingS
19781979
* @param id the identifier of the retention lease
19791980
* @param listener the callback when the retention lease is successfully removed and synced to replicas
19801981
*/
1981-
public void removeRetentionLease(final String id, final ActionListener<Void> listener) {
1982+
public void removeRetentionLease(final String id, final ActionListener<ReplicationResponse> listener) {
19821983
Objects.requireNonNull(listener);
19831984
assert assertPrimaryMode();
19841985
verifyNotClosed();
@@ -2020,7 +2021,7 @@ public void persistRetentionLeases() throws WriteStateException {
20202021
/**
20212022
* Syncs the current retention leases to all replicas.
20222023
*/
2023-
public void syncRetentionLeases(ActionListener<Void> listener) {
2024+
public void syncRetentionLeases(ActionListener<ReplicationResponse> listener) {
20242025
assert assertPrimaryMode();
20252026
verifyNotClosed();
20262027
final Tuple<Boolean, RetentionLeases> retentionLeases = getRetentionLeases(true);
@@ -2030,7 +2031,7 @@ public void syncRetentionLeases(ActionListener<Void> listener) {
20302031
shardId,
20312032
retentionLeases.v2(),
20322033
ActionListener.wrap(
2033-
r -> listener.onResponse(null),
2034+
listener::onResponse,
20342035
e -> {
20352036
logger.warn(new ParameterizedMessage(
20362037
"failed to sync retention leases [{}] after expiration check",
@@ -2042,7 +2043,7 @@ public void syncRetentionLeases(ActionListener<Void> listener) {
20422043
logger.trace("background syncing retention leases [{}] after expiration check", retentionLeases.v2());
20432044
retentionLeaseSyncer.backgroundSync(shardId, retentionLeases.v2(),
20442045
ActionListener.wrap(
2045-
r -> listener.onResponse(null),
2046+
listener::onResponse,
20462047
e -> {
20472048
logger.warn(new ParameterizedMessage(
20482049
"failed to background sync retention leases [{}] after expiration check",

‎server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.elasticsearch.ResourceAlreadyExistsException;
2727
import org.elasticsearch.Version;
2828
import org.elasticsearch.action.ActionListener;
29+
import org.elasticsearch.action.support.replication.ReplicationResponse;
2930
import org.elasticsearch.cluster.ClusterChangedEvent;
3031
import org.elasticsearch.cluster.ClusterState;
3132
import org.elasticsearch.cluster.ClusterStateApplier;
@@ -165,15 +166,15 @@ public IndicesClusterStateService(
165166
public void sync(
166167
final ShardId shardId,
167168
final RetentionLeases retentionLeases,
168-
final ActionListener<Void> listener) {
169+
final ActionListener<ReplicationResponse> listener) {
169170
Objects.requireNonNull(retentionLeaseSyncAction).sync(shardId, retentionLeases, listener);
170171
}
171172

172173
@Override
173174
public void backgroundSync(
174175
final ShardId shardId,
175176
final RetentionLeases retentionLeases,
176-
final ActionListener<Void> listener) {
177+
final ActionListener<ReplicationResponse> listener) {
177178
Objects.requireNonNull(retentionLeaseBackgroundSyncAction).backgroundSync(shardId, retentionLeases, listener);
178179
}
179180
});

‎server/src/test/java/org/elasticsearch/index/replication/RetentionLeasesReplicationTests.java‎

Lines changed: 30 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import org.elasticsearch.action.ActionListener;
2323
import org.elasticsearch.action.support.PlainActionFuture;
24+
import org.elasticsearch.action.support.replication.ReplicationResponse;
2425
import org.elasticsearch.cluster.metadata.IndexMetaData;
2526
import org.elasticsearch.common.Randomness;
2627
import org.elasticsearch.common.settings.Settings;
@@ -34,8 +35,6 @@
3435
import java.util.ArrayList;
3536
import java.util.List;
3637
import java.util.concurrent.CountDownLatch;
37-
import java.util.concurrent.atomic.AtomicReference;
38-
import java.util.function.Consumer;
3938

4039
import static org.hamcrest.Matchers.containsInAnyOrder;
4140
import static org.hamcrest.Matchers.equalTo;
@@ -76,20 +75,19 @@ public void testOutOfOrderRetentionLeasesRequests() throws Exception {
7675
Settings settings = Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build();
7776
int numberOfReplicas = between(1, 2);
7877
IndexMetaData indexMetaData = buildIndexMetaData(numberOfReplicas, settings, indexMapping);
79-
final List<RetentionLeaseSyncAction.Request> requests = new ArrayList<>();
8078
try (ReplicationGroup group = new ReplicationGroup(indexMetaData) {
8179
@Override
82-
protected void syncRetentionLeases(ShardId shardId, RetentionLeases leases, ActionListener<Void> listener) {
83-
requests.add(new RetentionLeaseSyncAction.Request(shardId, leases));
84-
listener.onResponse(null);
80+
protected void syncRetentionLeases(ShardId shardId, RetentionLeases leases, ActionListener<ReplicationResponse> listener) {
81+
listener.onResponse(new SyncRetentionLeasesResponse(new RetentionLeaseSyncAction.Request(shardId, leases)));
8582
}
8683
}) {
8784
group.startAll();
8885
int numLeases = between(1, 10);
86+
List<RetentionLeaseSyncAction.Request> requests = new ArrayList<>();
8987
for (int i = 0; i < numLeases; i++) {
90-
PlainActionFuture<Void> future = new PlainActionFuture<>();
88+
PlainActionFuture<ReplicationResponse> future = new PlainActionFuture<>();
9189
group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i, future);
92-
future.get();
90+
requests.add(((SyncRetentionLeasesResponse) future.actionGet()).syncRequest);
9391
}
9492
RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases();
9593
for (IndexShard replica : group.getReplicas()) {
@@ -104,56 +102,50 @@ public void testSyncRetentionLeasesWithPrimaryPromotion() throws Exception {
104102
Settings settings = Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build();
105103
int numberOfReplicas = between(2, 4);
106104
IndexMetaData indexMetaData = buildIndexMetaData(numberOfReplicas, settings, indexMapping);
107-
final AtomicReference<Consumer<RetentionLeaseSyncAction.Request>> onRequest = new AtomicReference<>();
108105
try (ReplicationGroup group = new ReplicationGroup(indexMetaData) {
109106
@Override
110-
protected void syncRetentionLeases(ShardId shardId, RetentionLeases leases, ActionListener<Void> listener) {
111-
onRequest.get().accept(new RetentionLeaseSyncAction.Request(shardId, leases));
112-
listener.onResponse(null);
107+
protected void syncRetentionLeases(ShardId shardId, RetentionLeases leases, ActionListener<ReplicationResponse> listener) {
108+
listener.onResponse(new SyncRetentionLeasesResponse(new RetentionLeaseSyncAction.Request(shardId, leases)));
113109
}
114110
}) {
115111
group.startAll();
116112
int numLeases = between(1, 100);
117-
final IndexShard newPrimary = randomFrom(group.getReplicas());
118-
final AtomicReference<RetentionLeases> latestRetentionLeasesOnNewPrimary = new AtomicReference<>(RetentionLeases.EMPTY);
119-
120-
onRequest.set(request -> {
121-
final RetentionLeases newRetentionLeases = request.getRetentionLeases();
113+
IndexShard newPrimary = randomFrom(group.getReplicas());
114+
RetentionLeases latestRetentionLeasesOnNewPrimary = RetentionLeases.EMPTY;
115+
for (int i = 0; i < numLeases; i++) {
116+
PlainActionFuture<ReplicationResponse> addLeaseFuture = new PlainActionFuture<>();
117+
group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i, addLeaseFuture);
118+
RetentionLeaseSyncAction.Request request = ((SyncRetentionLeasesResponse) addLeaseFuture.actionGet()).syncRequest;
122119
for (IndexShard replica : randomSubsetOf(group.getReplicas())) {
123120
group.executeRetentionLeasesSyncRequestOnReplica(request, replica);
124121
if (newPrimary == replica) {
125-
latestRetentionLeasesOnNewPrimary.updateAndGet(currentRetentionLeases
126-
-> newRetentionLeases.supersedes(currentRetentionLeases) ? newRetentionLeases : currentRetentionLeases);
122+
latestRetentionLeasesOnNewPrimary = request.getRetentionLeases();
127123
}
128124
}
129-
});
130-
131-
for (int i = 0; i < numLeases; i++) {
132-
PlainActionFuture<Void> addLeaseFuture = new PlainActionFuture<>();
133-
group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i, addLeaseFuture);
134-
addLeaseFuture.get();
135125
}
136126
group.promoteReplicaToPrimary(newPrimary).get();
137-
138127
// we need to make changes to retention leases to sync it to replicas
139128
// since we don't sync retention leases when promoting a new primary.
140-
141-
onRequest.set(request -> {
142-
for (IndexShard replica : group.getReplicas()) {
143-
group.executeRetentionLeasesSyncRequestOnReplica(request, replica);
144-
}
145-
});
146-
147-
final PlainActionFuture<Void> newLeaseFuture = new PlainActionFuture<>();
129+
PlainActionFuture<ReplicationResponse> newLeaseFuture = new PlainActionFuture<>();
148130
group.addRetentionLease("new-lease-after-promotion", randomNonNegativeLong(), "test", newLeaseFuture);
149-
final RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases();
131+
RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases();
150132
assertThat(leasesOnPrimary.primaryTerm(), equalTo(group.getPrimary().getOperationPrimaryTerm()));
151-
assertThat(leasesOnPrimary.version(), equalTo(latestRetentionLeasesOnNewPrimary.get().version() + 1L));
152-
assertThat(leasesOnPrimary.leases(), hasSize(latestRetentionLeasesOnNewPrimary.get().leases().size() + 1));
153-
newLeaseFuture.get();
133+
assertThat(leasesOnPrimary.version(), equalTo(latestRetentionLeasesOnNewPrimary.version() + 1L));
134+
assertThat(leasesOnPrimary.leases(), hasSize(latestRetentionLeasesOnNewPrimary.leases().size() + 1));
135+
RetentionLeaseSyncAction.Request request = ((SyncRetentionLeasesResponse) newLeaseFuture.actionGet()).syncRequest;
136+
for (IndexShard replica : group.getReplicas()) {
137+
group.executeRetentionLeasesSyncRequestOnReplica(request, replica);
138+
}
154139
for (IndexShard replica : group.getReplicas()) {
155140
assertThat(replica.getRetentionLeases(), equalTo(leasesOnPrimary));
156141
}
157142
}
158143
}
144+
145+
static final class SyncRetentionLeasesResponse extends ReplicationResponse {
146+
final RetentionLeaseSyncAction.Request syncRequest;
147+
SyncRetentionLeasesResponse(RetentionLeaseSyncAction.Request syncRequest) {
148+
this.syncRequest = syncRequest;
149+
}
150+
}
159151
}

‎server/src/test/java/org/elasticsearch/index/seqno/ReplicationTrackerTests.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
package org.elasticsearch.index.seqno;
2121

2222
import org.elasticsearch.action.ActionListener;
23+
import org.elasticsearch.action.support.replication.ReplicationResponse;
2324
import org.elasticsearch.cluster.routing.AllocationId;
2425
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
2526
import org.elasticsearch.cluster.routing.ShardRouting;
@@ -686,7 +687,7 @@ public void testPrimaryContextHandoff() throws IOException {
686687
final LongConsumer onUpdate = updatedGlobalCheckpoint -> {};
687688
final long primaryTerm = randomNonNegativeLong();
688689
final long globalCheckpoint = UNASSIGNED_SEQ_NO;
689-
final BiConsumer<RetentionLeases, ActionListener<Void>> onNewRetentionLease =
690+
final BiConsumer<RetentionLeases, ActionListener<ReplicationResponse>> onNewRetentionLease =
690691
(leases, listener) -> {};
691692
ReplicationTracker oldPrimary = new ReplicationTracker(
692693
shardId, aId.getId(), indexSettings, primaryTerm, globalCheckpoint, onUpdate, () -> 0L, onNewRetentionLease);

0 commit comments

Comments
 (0)