diff --git a/src/main/java/org/elasticsearch/Version.java b/src/main/java/org/elasticsearch/Version.java index c416e711151..e70797fa535 100644 --- a/src/main/java/org/elasticsearch/Version.java +++ b/src/main/java/org/elasticsearch/Version.java @@ -182,6 +182,8 @@ public class Version implements Serializable { public static final Version V_1_2_0 = new Version(V_1_2_0_ID, false, org.apache.lucene.util.Version.LUCENE_48); public static final int V_1_2_1_ID = /*00*/1020199; public static final Version V_1_2_1 = new Version(V_1_2_1_ID, false, org.apache.lucene.util.Version.LUCENE_48); + public static final int V_1_2_2_ID = /*00*/1020299; + public static final Version V_1_2_2 = new Version(V_1_2_2_ID, false, org.apache.lucene.util.Version.LUCENE_48); public static final int V_1_3_0_ID = /*00*/1030099; public static final Version V_1_3_0 = new Version(V_1_3_0_ID, false, org.apache.lucene.util.Version.LUCENE_48); public static final int V_2_0_0_ID = /*00*/2000099; @@ -203,6 +205,8 @@ public class Version implements Serializable { return V_2_0_0; case V_1_3_0_ID: return V_1_3_0; + case V_1_2_2_ID: + return V_1_2_2; case V_1_2_1_ID: return V_1_2_1; case V_1_2_0_ID: diff --git a/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java b/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java index 867a027d036..e68e6fb1082 100644 --- a/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java +++ b/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java @@ -20,6 +20,7 @@ package org.elasticsearch.indices.recovery; import com.google.common.collect.Maps; +import org.elasticsearch.Version; import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; @@ -29,7 +30,6 @@ import org.elasticsearch.transport.TransportRequest; import java.io.IOException; import java.util.Map; -import java.util.concurrent.atomic.AtomicLong; /** * @@ -115,6 +115,9 @@ public class StartRecoveryRequest extends TransportRequest { StoreFileMetaData md = StoreFileMetaData.readStoreFileMetaData(in); existingFiles.put(md.name(), md); } + if (in.getVersion().onOrAfter(Version.V_1_2_2)) { + recoveryType = RecoveryState.Type.fromId(in.readByte()); + } } @Override @@ -129,5 +132,8 @@ public class StartRecoveryRequest extends TransportRequest { for (StoreFileMetaData md : existingFiles.values()) { md.writeTo(out); } + if (out.getVersion().onOrAfter(Version.V_1_2_2)) { + out.writeByte(recoveryType.id()); + } } } diff --git a/src/test/java/org/elasticsearch/indices/recovery/StartRecoveryRequestTest.java b/src/test/java/org/elasticsearch/indices/recovery/StartRecoveryRequestTest.java new file mode 100644 index 00000000000..b1181c6a32a --- /dev/null +++ b/src/test/java/org/elasticsearch/indices/recovery/StartRecoveryRequestTest.java @@ -0,0 +1,82 @@ +/* + * Licensed to Elasticsearch under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.elasticsearch.indices.recovery; + +import org.elasticsearch.Version; +import org.elasticsearch.cluster.node.DiscoveryNode; +import org.elasticsearch.common.io.stream.InputStreamStreamInput; +import org.elasticsearch.common.io.stream.OutputStreamStreamOutput; +import org.elasticsearch.common.transport.LocalTransportAddress; +import org.elasticsearch.index.shard.ShardId; +import org.elasticsearch.index.store.StoreFileMetaData; +import org.elasticsearch.test.ElasticsearchTestCase; +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.util.Collections; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.nullValue; + +/** + */ +public class StartRecoveryRequestTest extends ElasticsearchTestCase { + + @Test + public void testSerialization() throws Exception { + Version targetNodeVersion = randomVersion(); + StartRecoveryRequest outRequest = new StartRecoveryRequest( + new ShardId("test", 0), + new DiscoveryNode("a", new LocalTransportAddress("1"), targetNodeVersion), + new DiscoveryNode("b", new LocalTransportAddress("1"), targetNodeVersion), + true, + Collections.emptyMap(), + RecoveryState.Type.RELOCATION, + 1l + + ); + ByteArrayOutputStream outBuffer = new ByteArrayOutputStream(); + OutputStreamStreamOutput out = new OutputStreamStreamOutput(outBuffer); + out.setVersion(targetNodeVersion); + outRequest.writeTo(out); + + ByteArrayInputStream inBuffer = new ByteArrayInputStream(outBuffer.toByteArray()); + InputStreamStreamInput in = new InputStreamStreamInput(inBuffer); + in.setVersion(targetNodeVersion); + StartRecoveryRequest inRequest = new StartRecoveryRequest(); + inRequest.readFrom(in); + + assertThat(outRequest.shardId(), equalTo(inRequest.shardId())); + assertThat(outRequest.sourceNode(), equalTo(inRequest.sourceNode())); + assertThat(outRequest.targetNode(), equalTo(inRequest.targetNode())); + assertThat(outRequest.markAsRelocated(), equalTo(inRequest.markAsRelocated())); + assertThat(outRequest.existingFiles(), equalTo(inRequest.existingFiles())); + assertThat(outRequest.recoveryId(), equalTo(inRequest.recoveryId())); + if (targetNodeVersion.onOrAfter(Version.V_1_2_2)) { + assertThat(outRequest.recoveryType(), equalTo(inRequest.recoveryType())); + } else { + assertThat(inRequest.recoveryType(), nullValue()); + } + } + + + +}