Tighten sequence number assertion

We have an assertion in the engine regarding the initial state of a
sequence number before an indexing operation. This assertion is too
loose, it catches operations during recovery from old indices where
sequence numbers do not even exist. This commit tightens these
assertions to not catch such operations and enables us to reenable some
tests.

Relates #21509

core/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -936,9 +936,6 @@ public abstract class Engine implements Closeable {
 
         public Operation(Term uid, long seqNo, long version, VersionType versionType, Origin origin, long startTime) {
             this.uid = uid;
-            // nocommit move these to InternalEngine where we can assert on the engine version
-            assert origin != Origin.PRIMARY || seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO : "seqNo should not be set when origin is PRIMARY";
-            assert origin == Origin.PRIMARY || seqNo >= 0 : "seqNo should be set when origin is not PRIMARY";
             this.seqNo = seqNo;
             this.version = version;
             this.versionType = versionType;

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

@@ -528,7 +528,21 @@ public class InternalEngine extends Engine {
         return false;
     }
 
+    private boolean assertSequenceNumber(final Engine.Operation.Origin origin, final long seqNo) {
+        if (engineConfig.getIndexSettings().getIndexVersionCreated().before(Version.V_6_0_0_alpha1) && origin == Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
+            // legacy support
+            return seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO;
+        } else if (origin == Operation.Origin.PRIMARY) {
+            // sequence number should not be set when operation origin is primary
+            return seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO;
+        } else {
+            // sequence number should be set when operation origin is not primary
+            return seqNo >= 0;
+        }
+    }
+
     private IndexResult innerIndex(Index index) throws IOException {
+        assert assertSequenceNumber(index.origin(), index.seqNo());
         final Translog.Location location;
         final long updatedVersion;
         IndexResult indexResult = null;
@@ -692,6 +706,7 @@ public class InternalEngine extends Engine {
     }
 
     private DeleteResult innerDelete(Delete delete) throws IOException {
+        assert assertSequenceNumber(delete.origin(), delete.seqNo());
         final Translog.Location location;
         final long updatedVersion;
         final boolean found;

core/src/test/java/org/elasticsearch/bwcompat/OldIndexBackwardsCompatibilityIT.java

@@ -91,9 +91,6 @@ import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 // needs at least 2 nodes since it bumps replicas to 1
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0)
 @LuceneTestCase.SuppressFileSystems("ExtrasFS")
-@LuceneTestCase.AwaitsFix(bugUrl = "needs a solution for translog operations that are recovered from the translog, don't have a seq no " +
-    "and trigger assertions in Engine.Operation")
-// nocommit. Fix ^^^ please.
 public class OldIndexBackwardsCompatibilityIT extends ESIntegTestCase {
     // TODO: test for proper exception on unsupported indexes (maybe via separate test?)
     // We have a 0.20.6.zip etc for this.

core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -166,7 +166,6 @@ import static org.hamcrest.Matchers.nullValue;
 
 public class InternalEngineTests extends ESTestCase {
 
-
     protected final ShardId shardId = new ShardId(new Index("index", "_na_"), 1);
     private static final IndexSettings INDEX_SETTINGS = IndexSettingsModule.newIndexSettings("index", Settings.EMPTY);
 
@@ -1207,7 +1206,7 @@ public class InternalEngineTests extends ESTestCase {
             assertThat(result.getFailure(), instanceOf(IllegalArgumentException.class));
             assertThat(result.getFailure().getMessage(), containsString("version type [FORCE] may not be used for non-translog operations"));
 
-            index = new Engine.Index(newUid("1"), doc, randomIntBetween(0, 16), 84, VersionType.FORCE,
+            index = new Engine.Index(newUid("1"), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 84, VersionType.FORCE,
                     Engine.Operation.Origin.LOCAL_TRANSLOG_RECOVERY, 0, -1, false);
             result = engine.index(index);
             assertThat(result.getVersion(), equalTo(84L));
@@ -2086,8 +2085,6 @@ public class InternalEngineTests extends ESTestCase {
         return new Mapping(Version.CURRENT, root, new MetadataFieldMapper[0], emptyMap());
     }
 
-    // nocommit
-    @AwaitsFix(bugUrl = "trips assertions in Engine, assertions should be moved to InternalEngine")
     public void testUpgradeOldIndex() throws IOException {
         List<Path> indexes = new ArrayList<>();
         try (DirectoryStream<Path> stream = Files.newDirectoryStream(getBwcIndicesPath(), "index-*.zip")) {