move primaryTerm to a long

Author: bleskes

core/src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.cluster.metadata;
 
-import com.carrotsearch.hppc.IntArrayList;
+import com.carrotsearch.hppc.LongArrayList;
 import com.carrotsearch.hppc.cursors.ObjectCursor;
 import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
 import org.elasticsearch.Version;
@@ -173,7 +173,7 @@ public static State fromString(String state) {
 
     private final String index;
     private final long version;
-    private final int[] primaryTerms;
+    private final long[] primaryTerms;
 
     private final State state;
 
@@ -195,7 +195,7 @@ public static State fromString(String state) {
     private final Version indexUpgradedVersion;
     private final org.apache.lucene.util.Version minimumCompatibleLuceneVersion;
 
-    private IndexMetaData(String index, long version, int[] primaryTerms, State state, Settings settings, ImmutableOpenMap<String, MappingMetaData> mappings, ImmutableOpenMap<String, AliasMetaData> aliases, ImmutableOpenMap<String, Custom> customs) {
+    private IndexMetaData(String index, long version, long[] primaryTerms, State state, Settings settings, ImmutableOpenMap<String, MappingMetaData> mappings, ImmutableOpenMap<String, AliasMetaData> aliases, ImmutableOpenMap<String, Custom> customs) {
         Integer maybeNumberOfShards = settings.getAsInt(SETTING_NUMBER_OF_SHARDS, null);
         if (maybeNumberOfShards == null) {
             throw new IllegalArgumentException("must specify numberOfShards for index [" + index + "]");
@@ -288,7 +288,7 @@ public long getVersion() {
      * a primary shard is assigned after a full cluster restart (see {@link ShardRouting#initialize(java.lang.String, long)}
      * or a replica shard is promoted to a primary (see {@link ShardRouting#moveToPrimary()}).
      **/
-    public int primaryTerm(int shardId) {
+    public long primaryTerm(int shardId) {
         return this.primaryTerms[shardId];
     }
 
@@ -471,7 +471,7 @@ private static class IndexMetaDataDiff implements Diff<IndexMetaData> {
 
         private final String index;
         private final long version;
-        private final int[] primaryTerms;
+        private final long[] primaryTerms;
         private final State state;
         private final Settings settings;
         private final Diff<ImmutableOpenMap<String, MappingMetaData>> mappings;
@@ -494,7 +494,7 @@ public IndexMetaDataDiff(StreamInput in) throws IOException {
             version = in.readLong();
             state = State.fromId(in.readByte());
             settings = Settings.readSettingsFromStream(in);
-            primaryTerms = in.readVIntArray();
+            primaryTerms = in.readVLongArray();
             mappings = DiffableUtils.readImmutableOpenMapDiff(in, MappingMetaData.PROTO);
             aliases = DiffableUtils.readImmutableOpenMapDiff(in, AliasMetaData.PROTO);
             customs = DiffableUtils.readImmutableOpenMapDiff(in, new DiffableUtils.KeyedReader<Custom>() {
@@ -516,7 +516,7 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeLong(version);
             out.writeByte(state.id);
             Settings.writeSettingsToStream(settings, out);
-            out.writeVIntArray(primaryTerms);
+            out.writeVLongArray(primaryTerms);
             mappings.writeTo(out);
             aliases.writeTo(out);
             customs.writeTo(out);
@@ -542,7 +542,7 @@ public IndexMetaData readFrom(StreamInput in) throws IOException {
         builder.version(in.readLong());
         builder.state(State.fromId(in.readByte()));
         builder.settings(readSettingsFromStream(in));
-        builder.primaryTerms(in.readVIntArray());
+        builder.primaryTerms(in.readVLongArray());
         int mappingsSize = in.readVInt();
         for (int i = 0; i < mappingsSize; i++) {
             MappingMetaData mappingMd = MappingMetaData.PROTO.readFrom(in);
@@ -568,7 +568,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeLong(version);
         out.writeByte(state.id());
         writeSettingsToStream(settings, out);
-        out.writeVIntArray(primaryTerms);
+        out.writeVLongArray(primaryTerms);
         out.writeVInt(mappings.size());
         for (ObjectCursor<MappingMetaData> cursor : mappings.values()) {
             cursor.value.writeTo(out);
@@ -597,7 +597,7 @@ public static class Builder {
         private String index;
         private State state = State.OPEN;
         private long version = 1;
-        private int[] primaryTerms = null;
+        private long[] primaryTerms = null;
         private Settings settings = Settings.Builder.EMPTY_SETTINGS;
         private final ImmutableOpenMap.Builder<String, MappingMetaData> mappings;
         private final ImmutableOpenMap.Builder<String, AliasMetaData> aliases;
@@ -739,7 +739,7 @@ public Builder version(long version) {
          * returns the primary term for the given shard.
          * See {@link IndexMetaData#primaryTerm(int)} for more information.
          */
-        public int primaryTerm(int shardId) {
+        public long primaryTerm(int shardId) {
             if (primaryTerms == null) {
                 initializePrimaryTerms();
             }
@@ -750,15 +750,15 @@ public int primaryTerm(int shardId) {
          * sets the primary term for the given shard.
          * See {@link IndexMetaData#primaryTerm(int)} for more information.
          */
-        public Builder primaryTerm(int shardId, int primaryTerm) {
+        public Builder primaryTerm(int shardId, long primaryTerm) {
             if (primaryTerms == null) {
                 initializePrimaryTerms();
             }
             this.primaryTerms[shardId] = primaryTerm;
             return this;
         }
 
-        private void primaryTerms(int[] primaryTerms) {
+        private void primaryTerms(long[] primaryTerms) {
             this.primaryTerms = primaryTerms;
         }
 
@@ -767,7 +767,7 @@ private void initializePrimaryTerms() {
             if (numberOfShards() < 0) {
                 throw new IllegalStateException("you must set the number of shards before setting/reading primary terms");
             }
-            primaryTerms = new int[numberOfShards()];
+            primaryTerms = new long[numberOfShards()];
         }
 
 
@@ -912,10 +912,10 @@ public static IndexMetaData fromXContent(XContentParser parser) throws IOExcepti
                             }
                         }
                     } else if (fieldEquals(Fields.PRIMARY_TERMS, currentFieldName)) {
-                        IntArrayList list = new IntArrayList();
+                        LongArrayList list = new LongArrayList();
                         while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
                             if (token == XContentParser.Token.VALUE_NUMBER) {
-                                list.add(parser.intValue());
+                                list.add(parser.longValue());
                             } else {
                                 throw new IllegalStateException("found a non-numeric value under [" + Fields.PRIMARY_TERMS.underscore() + "]");
                             }

core/src/main/java/org/elasticsearch/cluster/routing/IndexRoutingTable.java

@@ -425,7 +425,7 @@ private Builder initializeAsRestore(IndexMetaData indexMetaData, RestoreSource r
             for (int shardId = 0; shardId < indexMetaData.getNumberOfShards(); shardId++) {
                 IndexShardRoutingTable.Builder indexShardRoutingBuilder = new IndexShardRoutingTable.Builder(new ShardId(indexMetaData.getIndex(), shardId));
                 for (int i = 0; i <= indexMetaData.getNumberOfReplicas(); i++) {
-                    final int primaryTerm = indexMetaData.primaryTerm(shardId);
+                    final long primaryTerm = indexMetaData.primaryTerm(shardId);
                     if (asNew && ignoreShards.contains(shardId)) {
                         // This shards wasn't completely snapshotted - restore it as new shard
                         indexShardRoutingBuilder.addShard(ShardRouting.newUnassigned(index, shardId, null, primaryTerm, i == 0, unassignedInfo));
@@ -446,7 +446,7 @@ private Builder initializeEmpty(IndexMetaData indexMetaData, UnassignedInfo unas
                 throw new IllegalStateException("trying to initialize an index with fresh shards, but already has shards created");
             }
             for (int shardId = 0; shardId < indexMetaData.getNumberOfShards(); shardId++) {
-                final int primaryTerm = indexMetaData.primaryTerm(shardId);
+                final long primaryTerm = indexMetaData.primaryTerm(shardId);
                 IndexShardRoutingTable.Builder indexShardRoutingBuilder = new IndexShardRoutingTable.Builder(new ShardId(indexMetaData.getIndex(), shardId));
                 for (int i = 0; i <= indexMetaData.getNumberOfReplicas(); i++) {
                     indexShardRoutingBuilder.addShard(ShardRouting.newUnassigned(index, shardId, null,primaryTerm, i == 0, unassignedInfo));

core/src/main/java/org/elasticsearch/cluster/routing/IndexShardRoutingTable.java

@@ -124,11 +124,11 @@ public IndexShardRoutingTable normalizeVersionsAndPrimaryTerms() {
             return this;
         }
         long highestVersion = shards.get(0).version();
-        int highestPrimaryTerm = shards.get(0).primaryTerm();
+        long highestPrimaryTerm = shards.get(0).primaryTerm();
         boolean requiresNormalization = false;
         for (int i = 1; i < shards.size(); i++) {
             final long version = shards.get(i).version();
-            final int primaryTerm = shards.get(i).primaryTerm();
+            final long primaryTerm = shards.get(i).primaryTerm();
             if (highestVersion != version || highestPrimaryTerm != primaryTerm) {
                 requiresNormalization = true;
             }

core/src/main/java/org/elasticsearch/cluster/routing/ShardRouting.java

@@ -47,7 +47,7 @@ public final class ShardRouting implements Streamable, ToXContent {
     private String currentNodeId;
     private String relocatingNodeId;
     private boolean primary;
-    private int primaryTerm;
+    private long primaryTerm;
     private ShardRoutingState state;
     private long version;
     private RestoreSource restoreSource;
@@ -70,7 +70,7 @@ public ShardRouting(ShardRouting copy, long version) {
         this(copy, version, copy.primaryTerm());
     }
 
-    public ShardRouting(ShardRouting copy, long version, int primaryTerm) {
+    public ShardRouting(ShardRouting copy, long version, long primaryTerm) {
         this(copy.index(), copy.id(), copy.currentNodeId(), copy.relocatingNodeId(), copy.restoreSource(), primaryTerm, copy.primary(), copy.state(), version, copy.unassignedInfo(), copy.allocationId(), true, copy.getExpectedShardSize());
     }
 
@@ -79,7 +79,7 @@ public ShardRouting(ShardRouting copy, long version, int primaryTerm) {
      * by either this class or tests. Visible for testing.
      */
     ShardRouting(String index, int shardId, String currentNodeId,
-                 String relocatingNodeId, RestoreSource restoreSource, int primaryTerm, boolean primary, ShardRoutingState state, long version,
+                 String relocatingNodeId, RestoreSource restoreSource, long primaryTerm, boolean primary, ShardRoutingState state, long version,
                  UnassignedInfo unassignedInfo, AllocationId allocationId, boolean internal, long expectedShardSize) {
         this.index = index;
         this.shardId = shardId;
@@ -109,7 +109,7 @@ public ShardRouting(ShardRouting copy, long version, int primaryTerm) {
     /**
      * Creates a new unassigned shard.
      */
-    public static ShardRouting newUnassigned(String index, int shardId, RestoreSource restoreSource, int primaryTerm, boolean primary, UnassignedInfo unassignedInfo) {
+    public static ShardRouting newUnassigned(String index, int shardId, RestoreSource restoreSource, long primaryTerm, boolean primary, UnassignedInfo unassignedInfo) {
         return new ShardRouting(index, shardId, null, null, restoreSource, primaryTerm, primary, ShardRoutingState.UNASSIGNED, 0, unassignedInfo, null, true, UNAVAILABLE_EXPECTED_SHARD_SIZE);
     }
 
@@ -261,7 +261,7 @@ public boolean primary() {
      *
      * See {@link org.elasticsearch.cluster.metadata.IndexMetaData#primaryTerm(int)} for more info.
      */
-    public int primaryTerm() {
+    public long primaryTerm() {
         return this.primaryTerm;
     }
 
@@ -334,7 +334,7 @@ public void readFromThin(StreamInput in) throws IOException {
         }
 
         primary = in.readBoolean();
-        primaryTerm = in.readVInt();
+        primaryTerm = in.readVLong();
         state = ShardRoutingState.fromValue(in.readByte());
 
         restoreSource = RestoreSource.readOptionalRestoreSource(in);
@@ -380,7 +380,7 @@ public void writeToThin(StreamOutput out) throws IOException {
         }
 
         out.writeBoolean(primary);
-        out.writeVInt(primaryTerm);
+        out.writeVLong(primaryTerm);
         out.writeByte(state.value());
 
         if (restoreSource != null) {
@@ -684,7 +684,8 @@ public int hashCode() {
         result = 31 * result + (currentNodeId != null ? currentNodeId.hashCode() : 0);
         result = 31 * result + (relocatingNodeId != null ? relocatingNodeId.hashCode() : 0);
         result = 31 * result + (primary ? 1 : 0);
-        result = 31 * result + primaryTerm;
+        result = 31 * result + (int) (primaryTerm ^ (primaryTerm >>> 32));
+        ;
         result = 31 * result + (state != null ? state.hashCode() : 0);
         result = 31 * result + (int) (version ^ (version >>> 32));
         result = 31 * result + (restoreSource != null ? restoreSource.hashCode() : 0);

core/src/test/java/org/elasticsearch/cluster/metadata/ToAndFromJsonMetaDataTests.java

@@ -154,7 +154,7 @@ public void testSimpleJsonFromAndTo() throws IOException {
         MetaData parsedMetaData = MetaData.Builder.fromXContent(XContentFactory.xContent(XContentType.JSON).createParser(metaDataSource));
 
         IndexMetaData indexMetaData = parsedMetaData.index("test1");
-        assertThat(indexMetaData.primaryTerm(0), equalTo(1));
+        assertThat(indexMetaData.primaryTerm(0), equalTo(1l));
         assertThat(indexMetaData.getNumberOfShards(), equalTo(1));
         assertThat(indexMetaData.getNumberOfReplicas(), equalTo(2));
         assertThat(indexMetaData.getCreationDate(), equalTo(-1l));
@@ -164,8 +164,8 @@ public void testSimpleJsonFromAndTo() throws IOException {
         indexMetaData = parsedMetaData.index("test2");
         assertThat(indexMetaData.getNumberOfShards(), equalTo(2));
         assertThat(indexMetaData.getNumberOfReplicas(), equalTo(3));
-        assertThat(indexMetaData.primaryTerm(0), equalTo(2));
-        assertThat(indexMetaData.primaryTerm(1), equalTo(2));
+        assertThat(indexMetaData.primaryTerm(0), equalTo(2l));
+        assertThat(indexMetaData.primaryTerm(1), equalTo(2l));
         assertThat(indexMetaData.getCreationDate(), equalTo(-1l));
         assertThat(indexMetaData.getSettings().getAsMap().size(), equalTo(5));
         assertThat(indexMetaData.getSettings().get("setting1"), equalTo("value1"));

core/src/test/java/org/elasticsearch/cluster/routing/RoutingTableTests.java

@@ -56,7 +56,7 @@ public class RoutingTableTests extends ESAllocationTestCase {
             .build());
     private ClusterState clusterState;
 
-    private final Map<String, int[]> primaryTermsPerIndex = new HashMap<>();
+    private final Map<String, long[]> primaryTermsPerIndex = new HashMap<>();
     private final Map<String, long[]> versionsPerIndex = new HashMap<>();
 
     @Override
@@ -114,7 +114,7 @@ private void incrementVersion(String index, int shard) {
     }
 
     private void incrementPrimaryTerm(String index) {
-        final int[] primaryTerms = primaryTermsPerIndex.get(index);
+        final long[] primaryTerms = primaryTermsPerIndex.get(index);
         for (int i = 0; i < primaryTerms.length; i++) {
             primaryTerms[i]++;
         }
@@ -167,7 +167,7 @@ private void failSomePrimaries(String index) {
     }
 
     private IndexMetaData.Builder createIndexMetaData(String indexName) {
-        primaryTermsPerIndex.put(indexName, new int[numberOfShards]);
+        primaryTermsPerIndex.put(indexName, new long[numberOfShards]);
         final IndexMetaData.Builder builder = new IndexMetaData.Builder(indexName)
                 .settings(DEFAULT_SETTINGS)
                 .numberOfReplicas(this.numberOfReplicas)
@@ -185,7 +185,7 @@ private void assertAllVersionAndPrimaryTerm() {
 
     private void assertVersionAndPrimaryTerm(String index) {
         final long[] versions = versionsPerIndex.get(index);
-        final int[] terms = primaryTermsPerIndex.get(index);
+        final long[] terms = primaryTermsPerIndex.get(index);
         final IndexMetaData indexMetaData = clusterState.metaData().index(index);
         for (IndexShardRoutingTable shardRoutingTable : this.testRoutingTable.index(index)) {
             final int shard = shardRoutingTable.shardId().id();

core/src/test/java/org/elasticsearch/cluster/routing/ShardRoutingTests.java

@@ -35,7 +35,7 @@
 public class ShardRoutingTests extends ESTestCase {
 
     public void testFrozenAfterRead() throws IOException {
-        int term = randomInt(200);
+        long term = randomInt(200);
         ShardRouting routing = TestShardRouting.newShardRouting("foo", 1, "node_1", null, null, term, false, ShardRoutingState.INITIALIZING, 1);
         routing.moveToPrimary();
         assertTrue(routing.primary());
@@ -53,7 +53,7 @@ public void testFrozenAfterRead() throws IOException {
     }
 
     public void testPrimaryTermIncrementOnPromotion() {
-        int term = randomInt(200);
+        long term = randomInt(200);
         ShardRouting routing = TestShardRouting.newShardRouting("foo", 1, "node_1", null, null, term, false, ShardRoutingState.STARTED, 1);
         routing.moveToPrimary();
         assertTrue(routing.primary());
@@ -64,7 +64,7 @@ public void testPrimaryTermIncrementOnPromotion() {
     }
 
     public void testIsSameAllocation() {
-        int term = randomInt(200);
+        long term = randomInt(200);
         ShardRouting unassignedShard0 = TestShardRouting.newShardRouting("test", 0, null, term, false, ShardRoutingState.UNASSIGNED, 1);
         ShardRouting unassignedShard1 = TestShardRouting.newShardRouting("test", 1, null, term, false, ShardRoutingState.UNASSIGNED, 1);
         ShardRouting initializingShard0 = TestShardRouting.newShardRouting("test", 0, "1", term, randomBoolean(), ShardRoutingState.INITIALIZING, 1);