Added checks to ensure backward compatibility.

Signed-off-by: Ketan Verma <ketan9495@gmail.com>

Author: ketanv3

server/src/main/java/org/opensearch/cluster/metadata/ComposableIndexTemplate.java

@@ -33,6 +33,7 @@
 package org.opensearch.cluster.metadata;
 
 import org.opensearch.LegacyESVersion;
+import org.opensearch.Version;
 import org.opensearch.cluster.AbstractDiffable;
 import org.opensearch.cluster.Diff;
 import org.opensearch.cluster.metadata.DataStream.TimestampField;
@@ -289,7 +290,11 @@ public DataStreamTemplate(TimestampField timestampField) {
         }
 
         public DataStreamTemplate(StreamInput in) throws IOException {
-            this.timestampField = in.readOptionalWriteable(TimestampField::new);
+            if (in.getVersion().onOrAfter(Version.V_1_0_0)) {
+                this.timestampField = in.readOptionalWriteable(TimestampField::new);
+            } else {
+                this.timestampField = DataStreamFieldMapper.Defaults.TIMESTAMP_FIELD;
+            }
         }
 
         public TimestampField getTimestampField() {
@@ -312,7 +317,9 @@ MapperService.SINGLE_MAPPING_NAME, singletonMap(
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            out.writeOptionalWriteable(timestampField);
+            if (out.getVersion().onOrAfter(Version.V_1_0_0)) {
+                out.writeOptionalWriteable(timestampField);
+            }
         }
 
         @Override
@@ -323,6 +330,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                 .endObject();
         }
 
+        public static DataStreamTemplate fromXContent(XContentParser parser) {
+            return PARSER.apply(parser, null);
+        }
+
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;

server/src/test/java/org/opensearch/cluster/metadata/DataStreamTemplateTests.java

@@ -0,0 +1,57 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.cluster.metadata;
+
+import org.opensearch.Version;
+import org.opensearch.cluster.metadata.ComposableIndexTemplate.DataStreamTemplate;
+import org.opensearch.common.io.stream.BytesStreamOutput;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.Writeable;
+import org.opensearch.common.xcontent.XContentParser;
+import org.opensearch.test.AbstractSerializingTestCase;
+import org.opensearch.test.VersionUtils;
+
+import java.io.IOException;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class DataStreamTemplateTests extends AbstractSerializingTestCase<DataStreamTemplate> {
+
+    @Override
+    protected DataStreamTemplate doParseInstance(XContentParser parser) throws IOException {
+        return DataStreamTemplate.fromXContent(parser);
+    }
+
+    @Override
+    protected Writeable.Reader<DataStreamTemplate> instanceReader() {
+        return DataStreamTemplate::new;
+    }
+
+    @Override
+    protected DataStreamTemplate createTestInstance() {
+        return new DataStreamTemplate(new DataStream.TimestampField("timestamp_" + randomAlphaOfLength(5)));
+    }
+
+    public void testBackwardCompatibleSerialization() throws Exception {
+        Version version = VersionUtils.getPreviousVersion(Version.V_1_0_0);
+        BytesStreamOutput out = new BytesStreamOutput();
+        out.setVersion(version);
+
+        DataStreamTemplate outTemplate = new DataStreamTemplate();
+        outTemplate.writeTo(out);
+        assertThat(out.size(), equalTo(0));
+
+        StreamInput in = out.bytes().streamInput();
+        in.setVersion(version);
+        DataStreamTemplate inTemplate = new DataStreamTemplate(in);
+
+        assertThat(inTemplate, equalTo(outTemplate));
+    }
+
+}