Relax TranslogWriter#assertNoSeqNumberConflict (#31569)

If the recovery and indexing are concurrently happening, it's possible
for a replica to receive the same operation twice: one from the
replication, and the other from the recovery. However, these operations
are not identical because we don't store the versionType of operations
in the Lucene index.

The TranslogWriter#assertNoSeqNumberConflict assertion has been tripped
several times in the CCR branch since we use Lucene in peer-recovery.
This commit relaxes that assertion by excluding the versionType from the
check.

Author: dnhatn

server/src/main/java/org/elasticsearch/index/translog/TranslogWriter.java

@@ -204,14 +204,20 @@ private synchronized boolean assertNoSeqNumberConflict(long seqNo, BytesReferenc
                 Translog.Operation newOp = Translog.readOperation(new BufferedChecksumStreamInput(data.streamInput()));
                 Translog.Operation prvOp = Translog.readOperation(new BufferedChecksumStreamInput(previous.v1().streamInput()));
                 // TODO: We haven't had timestamp for Index operations in Lucene yet, we need to loosen this check without timestamp.
+                // We don't store versionType in Lucene index, we need to exclude it from this check
                 final boolean sameOp;
                 if (newOp instanceof Translog.Index && prvOp instanceof Translog.Index) {
                     final Translog.Index o1 = (Translog.Index) newOp;
                     final Translog.Index o2 = (Translog.Index) prvOp;
                     sameOp = Objects.equals(o1.id(), o2.id()) && Objects.equals(o1.type(), o2.type())
                         && Objects.equals(o1.source(), o2.source()) && Objects.equals(o1.routing(), o2.routing())
                         && o1.primaryTerm() == o2.primaryTerm() && o1.seqNo() == o2.seqNo()
-                        && o1.version() == o2.version() && o1.versionType() == o2.versionType();
+                        && o1.version() == o2.version();
+                } else if (newOp instanceof Translog.Delete && prvOp instanceof Translog.Delete) {
+                    final Translog.Delete o1 = (Translog.Delete) newOp;
+                    final Translog.Delete o2 = (Translog.Delete) prvOp;
+                    sameOp = Objects.equals(o1.id(), o2.id()) && Objects.equals(o1.type(), o2.type())
+                        && o1.primaryTerm() == o2.primaryTerm() && o1.seqNo() == o2.seqNo() && o1.version() == o2.version();
                 } else {
                     sameOp = false;
                 }

server/src/test/java/org/elasticsearch/versioning/SimpleVersioningIT.java

@@ -798,6 +798,9 @@ public void testSpecialVersioning() {
         IndexResponse doc3 = client().prepareIndex("test", "type", "1").setSource("field", "value3")
             .setVersion(Versions.MATCH_DELETED).setVersionType(VersionType.INTERNAL).execute().actionGet();
         assertThat(doc3.getVersion(), equalTo(3L));
+        IndexResponse doc4 = client().prepareIndex("test", "type", "1").setSource("field", "value4")
+            .setVersion(4L).setVersionType(VersionType.EXTERNAL_GTE).execute().actionGet();
+        assertThat(doc4.getVersion(), equalTo(4L));
         // Make sure that these versions are replicated correctly
         client().admin().indices().prepareUpdateSettings("test")
             .setSettings(Settings.builder().put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 1)).get();