Ignore Lucene index in peer recovery if translog corrupted

[dnhatn]

If the translog on a replica is corrupt, we should not perform an operation-based recovery or utilize sync_id as we won't be able to open an engine in the next step. This change adds an extra validation that ensures translog is okay when preparing a peer recovery request.

[elasticmachine]

Pinging @elastic/es-distributed (:Distributed/Recovery)

[dnhatn]

An alternative is to mark the store as corrupted if we failed to locally recover up to the global checkpoint. But it seems harsh, and we also need to handle a situation where we failed to mark the store as corrupted.

[dnhatn]

run elasticsearch-ci/1

[ywelsch]

LGTM

[henningandersen]

LGTM2

[dnhatn]

Thanks Yannick and Henning.

[hackerwin7]

before this patch, peer recovery maybe occur a exception like this:

Caused by: java.nio.file.NoSuchFileException: /data02/es_data/nodes/0/indices/2k_Oju9dRuqWx4tzbos53g/0/translog/translog.ckp
        at sun.nio.fs.UnixException.translateToIOException(UnixException.java:92) ~[?:?]
        at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:111) ~[?:?]
        at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:116) ~[?:?]
        at sun.nio.fs.UnixFileSystemProvider.newByteChannel(UnixFileSystemProvider.java:215) ~[?:?]
        at java.nio.file.Files.newByteChannel(Files.java:370) ~[?:?]
        at java.nio.file.Files.newByteChannel(Files.java:421) ~[?:?]
        at org.apache.lucene.store.SimpleFSDirectory.openInput(SimpleFSDirectory.java:77) ~[lucene-core-7.7.0.jar:7.7.0 8c831daf4eb41153c25ddb152501ab5bae3ea3d5 - jimczi - 2019-02-04 23:16:28]
        at org.elasticsearch.index.translog.Checkpoint.read(Checkpoint.java:188) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.index.translog.Translog.readCheckpoint(Translog.java:1847) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.index.translog.Translog.readCheckpoint(Translog.java:1865) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.index.translog.Translog.readGlobalCheckpoint(Translog.java:1860) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.index.shard.IndexShard.innerOpenEngineAndTranslog(IndexShard.java:1483) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.index.shard.IndexShard.openEngineAndSkipTranslogRecovery(IndexShard.java:1455) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.indices.recovery.RecoveryTarget.lambda$prepareForTranslogOperations$0(RecoveryTarget.java:302) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.action.ActionListener.completeWith(ActionListener.java:197) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.indices.recovery.RecoveryTarget.prepareForTranslogOperations(RecoveryTarget.java:297) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.indices.recovery.PeerRecoveryTargetService$PrepareForTranslogOperationsRequestHandler.messageReceived(PeerRecoveryTargetService.java:436) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.indices.recovery.PeerRecoveryTargetService$PrepareForTranslogOperationsRequestHandler.messageReceived(PeerRecoveryTargetService.java:430) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.transport.TransportRequestHandler.messageReceived(TransportRequestHandler.java:30) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.transport.RequestHandlerRegistry.processMessageReceived(RequestHandlerRegistry.java:66) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.transport.TcpTransport$RequestHandler.doRun(TcpTransport.java:1087) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:751) ~[elasticsearch-6.8.0.jar:6.8.0]
        at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37) ~[elasticsearch-6.8.0.jar:6.8.0]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
        at java.lang.Thread.run(Thread.java:834) ~[?:?]