honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove fileBasedRecovery flag (#45146) 

Today `RecoveryTarget#prepareForTranslogOperations` takes a boolean flag
indicating whether the recovery is file-based or not. This was used in 6.x to
bootstrap some commit data that were missing in indices created in 5.x:

b506955f8d/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java (L298-L300)

This flag no longer has any effect, so this commit removes it.

Backport of #45131 to 7.x.
This commit is contained in:
David Turner 2019-08-05 08:17:40 +01:00 committed by GitHub
parent 41815ed614
commit 13a167051f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 24 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java
View File

@ -367,7 +367,7 @@ public class PeerRecoveryTargetService implements IndexEventListener {
public void messageReceived(RecoveryPrepareForTranslogOperationsRequest request, TransportChannel channel, Task task) {
try (RecoveryRef recoveryRef = onGoingRecoveries.getRecoverySafe(request.recoveryId(), request.shardId())) {
final ActionListener<TransportResponse> listener = new ChannelActionListener<>(channel, Actions.PREPARE_TRANSLOG, request);
recoveryRef.target().prepareForTranslogOperations(request.isFileBasedRecovery(), request.totalTranslogOps(),
recoveryRef.target().prepareForTranslogOperations(request.totalTranslogOps(),
ActionListener.map(listener, nullVal -> TransportResponse.Empty.INSTANCE));
}
}

21
server/src/main/java/org/elasticsearch/indices/recovery/RecoveryPrepareForTranslogOperationsRequest.java
View File

@ -33,13 +33,11 @@ class RecoveryPrepareForTranslogOperationsRequest extends TransportRequest {
private final long recoveryId;
private final ShardId shardId;
private final int totalTranslogOps;
private final boolean fileBasedRecovery;
RecoveryPrepareForTranslogOperationsRequest(long recoveryId, ShardId shardId, int totalTranslogOps, boolean fileBasedRecovery) {
RecoveryPrepareForTranslogOperationsRequest(long recoveryId, ShardId shardId, int totalTranslogOps) {
this.recoveryId = recoveryId;
this.shardId = shardId;
this.totalTranslogOps = totalTranslogOps;
this.fileBasedRecovery = fileBasedRecovery;
}
RecoveryPrepareForTranslogOperationsRequest(StreamInput in) throws IOException {
@ -50,10 +48,8 @@ class RecoveryPrepareForTranslogOperationsRequest extends TransportRequest {
if (in.getVersion().before(Version.V_6_0_0_alpha1)) {
in.readLong(); // maxUnsafeAutoIdTimestamp
}
if (in.getVersion().onOrAfter(Version.V_6_2_0)) {
fileBasedRecovery = in.readBoolean();
} else {
fileBasedRecovery = true;
if (in.getVersion().onOrAfter(Version.V_6_2_0) && in.getVersion().before(Version.V_7_4_0)) {
in.readBoolean(); // was fileBasedRecovery
}
}
@ -69,13 +65,6 @@ class RecoveryPrepareForTranslogOperationsRequest extends TransportRequest {
return totalTranslogOps;
}
/**
* Whether or not the recovery is file based
*/
public boolean isFileBasedRecovery() {
return fileBasedRecovery;
}
@Override
public void writeTo(StreamOutput out) throws IOException {
super.writeTo(out);
@ -85,8 +74,8 @@ class RecoveryPrepareForTranslogOperationsRequest extends TransportRequest {
if (out.getVersion().before(Version.V_6_0_0_alpha1)) {
out.writeLong(IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP); // maxUnsafeAutoIdTimestamp
}
if (out.getVersion().onOrAfter(Version.V_6_2_0)) {
out.writeBoolean(fileBasedRecovery);
if (out.getVersion().onOrAfter(Version.V_6_2_0) && out.getVersion().before(Version.V_7_4_0)) {
out.writeBoolean(true); // was fileBasedRecovery
}
}
}

7
server/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java
View File

@ -285,8 +285,7 @@ public class RecoverySourceHandler {
sendFileStep.whenComplete(r -> {
assert Transports.assertNotTransportThread(RecoverySourceHandler.this + "[prepareTargetForTranslog]");
// For a sequence based recovery, the target can keep its local translog
prepareTargetForTranslog(isSequenceNumberBasedRecovery == false,
shard.estimateNumberOfHistoryOperations("peer-recovery", startingSeqNo), prepareEngineStep);
prepareTargetForTranslog(shard.estimateNumberOfHistoryOperations("peer-recovery", startingSeqNo), prepareEngineStep);
}, onFailure);
prepareEngineStep.whenComplete(prepareEngineTime -> {
@ -602,7 +601,7 @@ public class RecoverySourceHandler {
return true;
}
void prepareTargetForTranslog(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<TimeValue> listener) {
void prepareTargetForTranslog(int totalTranslogOps, ActionListener<TimeValue> listener) {
StopWatch stopWatch = new StopWatch().start();
final ActionListener<Void> wrappedListener = ActionListener.wrap(
nullVal -> {
@ -616,7 +615,7 @@ public class RecoverySourceHandler {
// garbage collection (not the JVM's GC!) of tombstone deletes.
logger.trace("recovery [phase1]: prepare remote engine for translog");
cancellableThreads.execute(() ->
recoveryTarget.prepareForTranslogOperations(fileBasedRecovery, totalTranslogOps, wrappedListener));
recoveryTarget.prepareForTranslogOperations(totalTranslogOps, wrappedListener));
}
/**

2
server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java
View File

@ -281,7 +281,7 @@ public class RecoveryTarget extends AbstractRefCounted implements RecoveryTarget
/*** Implementation of {@link RecoveryTargetHandler } */
@Override
public void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<Void> listener) {
public void prepareForTranslogOperations(int totalTranslogOps, ActionListener<Void> listener) {
ActionListener.completeWith(listener, () -> {
state().getTranslog().totalOperations(totalTranslogOps);
indexShard().openEngineAndSkipTranslogRecovery();

3
server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTargetHandler.java
View File

@ -33,10 +33,9 @@ public interface RecoveryTargetHandler {
/**
* Prepares the target to receive translog operations, after all file have been copied
*
* @param fileBasedRecovery whether or not this call is part of an file based recovery
* @param totalTranslogOps total translog operations expected to be sent
*/
void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<Void> listener);
void prepareForTranslogOperations(int totalTranslogOps, ActionListener<Void> listener);
/**
* The finalize request refreshes the engine now that new segments are available, enables garbage collection of tombstone files, updates

4
server/src/main/java/org/elasticsearch/indices/recovery/RemoteRecoveryTargetHandler.java
View File

@ -77,9 +77,9 @@ public class RemoteRecoveryTargetHandler implements RecoveryTargetHandler {
}
@Override
public void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<Void> listener) {
public void prepareForTranslogOperations(int totalTranslogOps, ActionListener<Void> listener) {
transportService.submitRequest(targetNode, PeerRecoveryTargetService.Actions.PREPARE_TRANSLOG,
new RecoveryPrepareForTranslogOperationsRequest(recoveryId, shardId, totalTranslogOps, fileBasedRecovery),
new RecoveryPrepareForTranslogOperationsRequest(recoveryId, shardId, totalTranslogOps),
TransportRequestOptions.builder().withTimeout(recoverySettings.internalActionTimeout()).build(),
new ActionListenerResponseHandler<>(ActionListener.map(listener, r -> null),
in -> TransportResponse.Empty.INSTANCE, ThreadPool.Names.GENERIC));

5
server/src/test/java/org/elasticsearch/index/replication/IndexLevelReplicationTests.java
View File

@ -207,14 +207,13 @@ public class IndexLevelReplicationTests extends ESIndexLevelReplicationTestCase
Future<Void> fut = shards.asyncRecoverReplica(replica,
(shard, node) -> new RecoveryTarget(shard, node, recoveryListener) {
@Override
public void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps,
ActionListener<Void> listener) {
public void prepareForTranslogOperations(int totalTranslogOps, ActionListener<Void> listener) {
try {
indexedOnPrimary.await();
} catch (InterruptedException e) {
throw new AssertionError(e);
}
super.prepareForTranslogOperations(fileBasedRecovery, totalTranslogOps, listener);
super.prepareForTranslogOperations(totalTranslogOps, listener);
}
});
fut.get();

4
server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java
View File

@ -2734,8 +2734,8 @@ public class IndexShardTests extends IndexShardTestCase {
new RecoveryTarget(shard, discoveryNode, recoveryListener) {
// we're only checking that listeners are called when the engine is open, before there is no point
@Override
public void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<Void> listener) {
super.prepareForTranslogOperations(fileBasedRecovery, totalTranslogOps,
public void prepareForTranslogOperations(int totalTranslogOps, ActionListener<Void> listener) {
super.prepareForTranslogOperations(totalTranslogOps,
ActionListener.wrap(
r -> {
assertListenerCalled.accept(replica);

6
server/src/test/java/org/elasticsearch/indices/recovery/RecoverySourceHandlerTests.java
View File

@ -475,9 +475,9 @@ public class RecoverySourceHandlerTests extends ESTestCase {
}
@Override
void prepareTargetForTranslog(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<TimeValue> listener) {
void prepareTargetForTranslog(int totalTranslogOps, ActionListener<TimeValue> listener) {
prepareTargetForTranslogCalled.set(true);
super.prepareTargetForTranslog(fileBasedRecovery, totalTranslogOps, listener);
super.prepareTargetForTranslog(totalTranslogOps, listener);
}
@Override
@ -778,7 +778,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
class TestRecoveryTargetHandler implements RecoveryTargetHandler {
@Override
public void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<Void> listener) {
public void prepareForTranslogOperations(int totalTranslogOps, ActionListener<Void> listener) {
}
@Override

4
server/src/test/java/org/elasticsearch/indices/recovery/RecoveryTests.java
View File

@ -340,8 +340,8 @@ public class RecoveryTests extends ESIndexLevelReplicationTestCase {
updateMappings(replicaShard, primaryShard.indexSettings().getIndexMetaData());
recoverReplica(replicaShard, primaryShard, (r, sourceNode) -> new RecoveryTarget(r, sourceNode, recoveryListener) {
@Override
public void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<Void> listener) {
super.prepareForTranslogOperations(fileBasedRecovery, totalTranslogOps, listener);
public void prepareForTranslogOperations(int totalTranslogOps, ActionListener<Void> listener) {
super.prepareForTranslogOperations(totalTranslogOps, listener);
assertThat(replicaShard.getLastKnownGlobalCheckpoint(), equalTo(primaryShard.getLastKnownGlobalCheckpoint()));
}
@Override

4
test/framework/src/main/java/org/elasticsearch/indices/recovery/AsyncRecoveryTarget.java
View File

@ -45,8 +45,8 @@ public class AsyncRecoveryTarget implements RecoveryTargetHandler {
}
@Override
public void prepareForTranslogOperations(boolean fileBasedRecovery, int totalTranslogOps, ActionListener<Void> listener) {
executor.execute(() -> target.prepareForTranslogOperations(fileBasedRecovery, totalTranslogOps, listener));
public void prepareForTranslogOperations(int totalTranslogOps, ActionListener<Void> listener) {
executor.execute(() -> target.prepareForTranslogOperations(totalTranslogOps, listener));
}
@Override