type and id are lost upon serialization of Translog.Delete. (#24586)

This was introduced in #24460: the constructor of `Translog.Delete` that takes
a `StreamInput` does not set the type and id. To make it a bit more robust, I
made fields final so that forgetting to set them would make the compiler
complain.

Author: jpountz

core/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -41,6 +41,7 @@
 import org.elasticsearch.common.util.concurrent.ReleasableLock;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.index.mapper.Uid;
 import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.shard.IndexShardComponent;
 
@@ -54,6 +55,7 @@
 import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -975,17 +977,29 @@ public long getAutoGeneratedIdTimestamp() {
     }
 
     public static class Delete implements Operation {
-        public static final int SERIALIZATION_FORMAT = 2; // since 2.0-beta1 and 1.1
+        public static final int FORMAT_5_0 = 2; // 5.0 - 5.5
+        private static final int FORMAT_SINGLE_TYPE = FORMAT_5_0 + 1; // 5.5 - 6.0
 
-        private String type, id;
+        private final String type, id;
         private final Term uid;
         private final long version;
         private final VersionType versionType;
 
         public Delete(StreamInput in) throws IOException {
             final int format = in.readVInt();// SERIALIZATION_FORMAT
-            assert format == SERIALIZATION_FORMAT : "format was: " + format;
-            uid = new Term(in.readString(), in.readString());
+            assert format >= FORMAT_5_0 && format <= FORMAT_SINGLE_TYPE : "format was: " + format;
+            if (format >= FORMAT_SINGLE_TYPE) {
+                type = in.readString();
+                id = in.readString();
+                uid = new Term(in.readString(), in.readString());
+            } else {
+                uid = new Term(in.readString(), in.readString());
+                // the uid was constructed from the type and id so we can
+                // extract them back
+                Uid uidObject = Uid.createUid(uid.text());
+                type = uidObject.type();
+                id = uidObject.id();
+            }
             this.version = in.readLong();
             this.versionType = VersionType.fromValue(in.readByte());
             assert versionType.validateVersionForWrites(this.version);
@@ -1000,8 +1014,8 @@ public Delete(String type, String id, Term uid) {
         }
 
         public Delete(String type, String id, Term uid, long version, VersionType versionType) {
-            this.type = type;
-            this.id = id;
+            this.type = Objects.requireNonNull(type);
+            this.id = Objects.requireNonNull(id);
             this.uid = uid;
             this.version = version;
             this.versionType = versionType;
@@ -1044,7 +1058,9 @@ public Source getSource() {
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            out.writeVInt(SERIALIZATION_FORMAT);
+            out.writeVInt(FORMAT_SINGLE_TYPE);
+            out.writeString(type);
+            out.writeString(id);
             out.writeString(uid.field());
             out.writeString(uid.text());
             out.writeLong(version);

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

@@ -1720,20 +1720,20 @@ public void testBasicCreatedFlag() throws IOException {
         indexResult = engine.index(index);
         assertFalse(indexResult.isCreated());
 
-        engine.delete(new Engine.Delete(null, "1", newUid(doc)));
+        engine.delete(new Engine.Delete("doc", "1", newUid(doc)));
 
         index = indexForDoc(doc);
         indexResult = engine.index(index);
         assertTrue(indexResult.isCreated());
     }
 
     public void testCreatedFlagAfterFlush() throws IOException {
-        ParsedDocument doc = testParsedDocument("1", "test", null, -1, -1, testDocument(), B_1, null);
+        ParsedDocument doc = testParsedDocument("1", "doc", null, -1, -1, testDocument(), B_1, null);
         Engine.Index index = indexForDoc(doc);
         Engine.IndexResult indexResult = engine.index(index);
         assertTrue(indexResult.isCreated());
 
-        engine.delete(new Engine.Delete(null, "1", newUid(doc)));
+        engine.delete(new Engine.Delete("doc", "1", newUid(doc)));
 
         engine.flush();
 

core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java

@@ -50,6 +50,7 @@
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.mapper.ParsedDocument;
 import org.elasticsearch.index.mapper.Uid;
+import org.elasticsearch.index.mapper.UidFieldMapper;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog.Location;
 import org.elasticsearch.test.ESTestCase;
@@ -298,21 +299,21 @@ public void testStats() throws IOException {
         assertThat(stats.estimatedNumberOfOperations(), equalTo(0L));
         assertThat(stats.getTranslogSizeInBytes(), equalTo(firstOperationPosition));
         assertEquals(6, total.estimatedNumberOfOperations());
-        assertEquals(455, total.getTranslogSizeInBytes());
+        assertEquals(476, total.getTranslogSizeInBytes());
 
         BytesStreamOutput out = new BytesStreamOutput();
         total.writeTo(out);
         TranslogStats copy = new TranslogStats();
         copy.readFrom(out.bytes().streamInput());
 
         assertEquals(6, copy.estimatedNumberOfOperations());
-        assertEquals(455, copy.getTranslogSizeInBytes());
+        assertEquals(476, copy.getTranslogSizeInBytes());
 
         try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
             builder.startObject();
             copy.toXContent(builder, ToXContent.EMPTY_PARAMS);
             builder.endObject();
-            assertEquals("{\"translog\":{\"operations\":6,\"size_in_bytes\":455}}", builder.string());
+            assertEquals("{\"translog\":{\"operations\":6,\"size_in_bytes\":476}}", builder.string());
         }
 
         try {
@@ -1925,4 +1926,20 @@ public void testPendingDelete() throws IOException {
     public static Translog.Location randomTranslogLocation() {
         return new Translog.Location(randomLong(), randomLong(), randomInt());
     }
+
+    public void testSerialization() throws Exception {
+        // simulate legacy delete serialization
+        BytesStreamOutput out = new BytesStreamOutput();
+        out.writeVInt(Translog.Delete.FORMAT_5_0);
+        out.writeString(UidFieldMapper.NAME);
+        out.writeString("my_type#my_id");
+        out.writeLong(3); // version
+        out.writeByte(VersionType.INTERNAL.getValue());
+        out.writeLong(2); // seq no
+        out.writeLong(0); // primary term
+        StreamInput in = out.bytes().streamInput();
+        Translog.Delete serializedDelete = new Translog.Delete(in);
+        assertEquals("my_type", serializedDelete.type());
+        assertEquals("my_id", serializedDelete.id());
+    }
 }