Add NoOps to Lucene for failed delete ops (elastic#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 elastic#29530

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

@@ -842,7 +842,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();
@@ -1177,8 +1177,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

@@ -342,7 +342,6 @@ public void testReplicaOperationWithConcurrentPrimaryPromotion() throws Exceptio
      * test document failures (failures after seq_no generation) are added as noop operation to the translog
      * for primary and replica shards
      */
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/pull/33217")
     public void testDocumentFailureReplication() throws Exception {
         final IOException indexException = new IOException("simulated indexing failure");
         final IOException deleteException = new IOException("simulated deleting failure");
@@ -405,6 +404,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));
@@ -419,6 +421,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);
         }