Add NoOps to Lucene for failed delete ops (#33217)

Today we add a NoOp to Lucene and translog if we fail to process an
indexing operation. However, we are only adding NoOps to translog for
delete operations. In order to have a complete history in Lucene, we
should add NoOps of failed delete operations to both Lucene and translog.

Relates #29530

Author: dnhatn

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -898,7 +898,7 @@ public IndexResult index(Index index) throws IOException {
                     if (indexResult.getResultType() == Result.Type.SUCCESS) {
                         location = translog.add(new Translog.Index(index, indexResult));
                     } else if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
-                        // if we have document failure, record it as a no-op in the translog with the generated seq_no
+                        // if we have document failure, record it as a no-op in the translog and Lucene with the generated seq_no
                         final NoOp noOp = new NoOp(indexResult.getSeqNo(), index.primaryTerm(), index.origin(),
                             index.startTime(), indexResult.getFailure().toString());
                         location = innerNoOp(noOp).getTranslogLocation();
@@ -1249,8 +1249,10 @@ public DeleteResult delete(Delete delete) throws IOException {
                 if (deleteResult.getResultType() == Result.Type.SUCCESS) {
                     location = translog.add(new Translog.Delete(delete, deleteResult));
                 } else if (deleteResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
-                    location = translog.add(new Translog.NoOp(deleteResult.getSeqNo(),
-                            delete.primaryTerm(), deleteResult.getFailure().toString()));
+                    // if we have document failure, record it as a no-op in the translog and Lucene with the generated seq_no
+                    final NoOp noOp = new NoOp(deleteResult.getSeqNo(), delete.primaryTerm(), delete.origin(),
+                        delete.startTime(), deleteResult.getFailure().toString());
+                    location = innerNoOp(noOp).getTranslogLocation();
                 } else {
                     location = null;
                 }

server/src/test/java/org/elasticsearch/index/replication/IndexLevelReplicationTests.java

@@ -407,6 +407,9 @@ public long softUpdateDocument(Term term, Iterable<? extends IndexableField> doc
                 try (Translog.Snapshot snapshot = getTranslog(shard).newSnapshot()) {
                     assertThat(snapshot, SnapshotMatchers.containsOperationsInAnyOrder(expectedTranslogOps));
                 }
+                try (Translog.Snapshot snapshot = shard.getHistoryOperations("test", 0)) {
+                    assertThat(snapshot, SnapshotMatchers.containsOperationsInAnyOrder(expectedTranslogOps));
+                }
             }
             // unlike previous failures, these two failures replicated directly from the replication channel.
             indexResp = shards.index(new IndexRequest(index.getName(), "type", "any").source("{}", XContentType.JSON));
@@ -421,6 +424,9 @@ public long softUpdateDocument(Term term, Iterable<? extends IndexableField> doc
                 try (Translog.Snapshot snapshot = getTranslog(shard).newSnapshot()) {
                     assertThat(snapshot, SnapshotMatchers.containsOperationsInAnyOrder(expectedTranslogOps));
                 }
+                try (Translog.Snapshot snapshot = shard.getHistoryOperations("test", 0)) {
+                    assertThat(snapshot, SnapshotMatchers.containsOperationsInAnyOrder(expectedTranslogOps));
+                }
             }
             shards.assertAllEqual(1);
         }