Decouple TimeValue from Elasticsearch server classes (#29454)

* Decouple TimeValue from Elasticsearch server classes

This commit decouples the `TimeValue` class from the other server classes. This
is in preperation to move `TimeValue` into the `elasticsearch-core` jar,
allowing us to use it from projects that cannot depend on the elasticsearch-core
library.

Relates to #28504

server/src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthRequest.java

@@ -64,7 +64,7 @@ public class ClusterHealthRequest extends MasterNodeReadRequest<ClusterHealthReq
                 indices[i] = in.readString();
             }
         }
-        timeout = new TimeValue(in);
+        timeout = in.readTimeValue();
         if (in.readBoolean()) {
             waitForStatus = ClusterHealthStatus.fromValue(in.readByte());
         }
@@ -90,7 +90,7 @@ public class ClusterHealthRequest extends MasterNodeReadRequest<ClusterHealthReq
                 out.writeString(index);
             }
         }
-        timeout.writeTo(out);
+        out.writeTimeValue(timeout);
         if (waitForStatus == null) {
             out.writeBoolean(false);
         } else {

server/src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthResponse.java

@@ -184,7 +184,7 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
         timedOut = in.readBoolean();
         numberOfInFlightFetch = in.readInt();
         delayedUnassignedShards= in.readInt();
-        taskMaxWaitingTime = new TimeValue(in);
+        taskMaxWaitingTime = in.readTimeValue();
     }
 
     @Override
@@ -197,7 +197,7 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
         out.writeBoolean(timedOut);
         out.writeInt(numberOfInFlightFetch);
         out.writeInt(delayedUnassignedShards);
-        taskMaxWaitingTime.writeTo(out);
+        out.writeTimeValue(taskMaxWaitingTime);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/NodesHotThreadsRequest.java

@@ -99,7 +99,7 @@ public class NodesHotThreadsRequest extends BaseNodesRequest<NodesHotThreadsRequ
         threads = in.readInt();
         ignoreIdleThreads = in.readBoolean();
         type = in.readString();
-        interval = new TimeValue(in);
+        interval = in.readTimeValue();
         snapshots = in.readInt();
     }
 
@@ -109,7 +109,7 @@ public class NodesHotThreadsRequest extends BaseNodesRequest<NodesHotThreadsRequ
         out.writeInt(threads);
         out.writeBoolean(ignoreIdleThreads);
         out.writeString(type);
-        interval.writeTo(out);
+        out.writeTimeValue(interval);
         out.writeInt(snapshots);
     }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/GetTaskRequest.java

@@ -105,7 +105,7 @@ public class GetTaskRequest extends ActionRequest {
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         taskId = TaskId.readFromStream(in);
-        timeout = in.readOptionalWriteable(TimeValue::new);
+        timeout = in.readOptionalTimeValue();
         waitForCompletion = in.readBoolean();
     }
 
@@ -113,7 +113,7 @@ public class GetTaskRequest extends ActionRequest {
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         taskId.writeTo(out);
-        out.writeOptionalWriteable(timeout);
+        out.writeOptionalTimeValue(timeout);
         out.writeBoolean(waitForCompletion);
     }
 }

server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java

@@ -588,7 +588,7 @@ public class BulkRequest extends ActionRequest implements CompositeIndicesReques
             requests.add(DocWriteRequest.readDocumentRequest(in));
         }
         refreshPolicy = RefreshPolicy.readFrom(in);
-        timeout = new TimeValue(in);
+        timeout = in.readTimeValue();
     }
 
     @Override
@@ -600,7 +600,7 @@ public class BulkRequest extends ActionRequest implements CompositeIndicesReques
             DocWriteRequest.writeDocumentRequest(out, request);
         }
         refreshPolicy.writeTo(out);
-        timeout.writeTo(out);
+        out.writeTimeValue(timeout);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -511,7 +511,7 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
         }
         if (in.getVersion().before(Version.V_6_0_0_alpha1)) {
             in.readOptionalString(); // timestamp
-            in.readOptionalWriteable(TimeValue::new); // ttl
+            in.readOptionalTimeValue(); // ttl
         }
         source = in.readBytesReference();
         opType = OpType.fromId(in.readByte());

server/src/main/java/org/elasticsearch/action/support/master/AcknowledgedRequest.java

@@ -78,13 +78,13 @@ public abstract class AcknowledgedRequest<Request extends MasterNodeRequest<Requ
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        timeout = new TimeValue(in);
+        timeout = in.readTimeValue();
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        timeout.writeTo(out);
+        out.writeTimeValue(timeout);
     }
 
 }

server/src/main/java/org/elasticsearch/action/support/master/MasterNodeRequest.java

@@ -40,13 +40,13 @@ public abstract class MasterNodeRequest<Request extends MasterNodeRequest<Reques
 
     protected MasterNodeRequest(StreamInput in) throws IOException {
         super(in);
-        masterNodeTimeout = new TimeValue(in);
+        masterNodeTimeout = in.readTimeValue();
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        masterNodeTimeout.writeTo(out);
+        out.writeTimeValue(masterNodeTimeout);
     }
 
     /**
@@ -74,6 +74,6 @@ public abstract class MasterNodeRequest<Request extends MasterNodeRequest<Reques
         // TODO(talevy): throw exception once all MasterNodeRequest
         //               subclasses have been migrated to Writeable Readers
         super.readFrom(in);
-        masterNodeTimeout = new TimeValue(in);
+        masterNodeTimeout = in.readTimeValue();
     }
 }

server/src/main/java/org/elasticsearch/action/support/nodes/BaseNodesRequest.java

@@ -107,7 +107,7 @@ public abstract class BaseNodesRequest<Request extends BaseNodesRequest<Request>
         super.readFrom(in);
         nodesIds = in.readStringArray();
         concreteNodes = in.readOptionalArray(DiscoveryNode::new, DiscoveryNode[]::new);
-        timeout = in.readOptionalWriteable(TimeValue::new);
+        timeout = in.readOptionalTimeValue();
     }
 
     @Override
@@ -115,6 +115,6 @@ public abstract class BaseNodesRequest<Request extends BaseNodesRequest<Request>
         super.writeTo(out);
         out.writeStringArrayNullable(nodesIds);
         out.writeOptionalArray(concreteNodes);
-        out.writeOptionalWriteable(timeout);
+        out.writeOptionalTimeValue(timeout);
     }
 }

server/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java

@@ -187,7 +187,7 @@ public abstract class ReplicationRequest<Request extends ReplicationRequest<Requ
             shardId = null;
         }
         waitForActiveShards = ActiveShardCount.readFrom(in);
-        timeout = new TimeValue(in);
+        timeout = in.readTimeValue();
         index = in.readString();
         routedBasedOnClusterVersion = in.readVLong();
     }
@@ -202,7 +202,7 @@ public abstract class ReplicationRequest<Request extends ReplicationRequest<Requ
             out.writeBoolean(false);
         }
         waitForActiveShards.writeTo(out);
-        timeout.writeTo(out);
+        out.writeTimeValue(timeout);
         out.writeString(index);
         out.writeVLong(routedBasedOnClusterVersion);
     }

server/src/main/java/org/elasticsearch/action/support/single/instance/InstanceShardOperationRequest.java

@@ -118,7 +118,7 @@ public abstract class InstanceShardOperationRequest<Request extends InstanceShar
         } else {
             shardId = null;
         }
-        timeout = new TimeValue(in);
+        timeout = in.readTimeValue();
         concreteIndex = in.readOptionalString();
     }
 
@@ -127,7 +127,7 @@ public abstract class InstanceShardOperationRequest<Request extends InstanceShar
         super.writeTo(out);
         out.writeString(index);
         out.writeOptionalStreamable(shardId);
-        timeout.writeTo(out);
+        out.writeTimeValue(timeout);
         out.writeOptionalString(concreteIndex);
     }
 

server/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java

@@ -144,7 +144,7 @@ public class BaseTasksRequest<Request extends BaseTasksRequest<Request>> extends
         parentTaskId = TaskId.readFromStream(in);
         nodes = in.readStringArray();
         actions = in.readStringArray();
-        timeout = in.readOptionalWriteable(TimeValue::new);
+        timeout = in.readOptionalTimeValue();
     }
 
     @Override
@@ -154,7 +154,7 @@ public class BaseTasksRequest<Request extends BaseTasksRequest<Request>> extends
         parentTaskId.writeTo(out);
         out.writeStringArrayNullable(nodes);
         out.writeStringArrayNullable(actions);
-        out.writeOptionalWriteable(timeout);
+        out.writeOptionalTimeValue(timeout);
     }
 
     public boolean match(Task task) {

server/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -36,6 +36,7 @@ import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.geo.GeoPoint;
 import org.elasticsearch.common.text.Text;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
@@ -65,6 +66,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.function.IntFunction;
 import java.util.function.Supplier;
 
@@ -82,6 +84,26 @@ import static org.elasticsearch.ElasticsearchException.readStackTrace;
  * on {@link StreamInput}.
  */
 public abstract class StreamInput extends InputStream {
+
+    private static final Map<Byte, TimeUnit> BYTE_TIME_UNIT_MAP;
+
+    static {
+        final Map<Byte, TimeUnit> byteTimeUnitMap = new HashMap<>();
+        byteTimeUnitMap.put((byte)0, TimeUnit.NANOSECONDS);
+        byteTimeUnitMap.put((byte)1, TimeUnit.MICROSECONDS);
+        byteTimeUnitMap.put((byte)2, TimeUnit.MILLISECONDS);
+        byteTimeUnitMap.put((byte)3, TimeUnit.SECONDS);
+        byteTimeUnitMap.put((byte)4, TimeUnit.MINUTES);
+        byteTimeUnitMap.put((byte)5, TimeUnit.HOURS);
+        byteTimeUnitMap.put((byte)6, TimeUnit.DAYS);
+
+        for (TimeUnit value : TimeUnit.values()) {
+            assert byteTimeUnitMap.containsValue(value) : value;
+        }
+
+        BYTE_TIME_UNIT_MAP = Collections.unmodifiableMap(byteTimeUnitMap);
+    }
+
     private Version version = Version.CURRENT;
 
     /**
@@ -971,4 +993,24 @@ public abstract class StreamInput extends InputStream {
      * be a no-op depending on the underlying implementation if the information of the remaining bytes is not present.
      */
     protected abstract void ensureCanReadBytes(int length) throws EOFException;
+
+    /**
+     * Read a {@link TimeValue} from the stream
+     */
+    public TimeValue readTimeValue() throws IOException {
+        long duration = readZLong();
+        TimeUnit timeUnit = BYTE_TIME_UNIT_MAP.get(readByte());
+        return new TimeValue(duration, timeUnit);
+    }
+
+    /**
+     * Read an optional {@link TimeValue} from the stream, returning null if no TimeValue was written.
+     */
+    public @Nullable TimeValue readOptionalTimeValue() throws IOException {
+        if (readBoolean()) {
+            return readTimeValue();
+        } else {
+            return null;
+        }
+    }
 }

server/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -35,6 +35,7 @@ import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.geo.GeoPoint;
 import org.elasticsearch.common.io.stream.Writeable.Writer;
 import org.elasticsearch.common.text.Text;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableInstant;
@@ -54,11 +55,13 @@ import java.nio.file.NotDirectoryException;
 import java.time.ZonedDateTime;
 import java.util.Collections;
 import java.util.Date;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.function.IntFunction;
 
 /**
@@ -74,6 +77,25 @@ import java.util.function.IntFunction;
  */
 public abstract class StreamOutput extends OutputStream {
 
+    private static final Map<TimeUnit, Byte> TIME_UNIT_BYTE_MAP;
+
+    static {
+        final Map<TimeUnit, Byte> timeUnitByteMap = new EnumMap<>(TimeUnit.class);
+        timeUnitByteMap.put(TimeUnit.NANOSECONDS, (byte)0);
+        timeUnitByteMap.put(TimeUnit.MICROSECONDS, (byte)1);
+        timeUnitByteMap.put(TimeUnit.MILLISECONDS, (byte)2);
+        timeUnitByteMap.put(TimeUnit.SECONDS, (byte)3);
+        timeUnitByteMap.put(TimeUnit.MINUTES, (byte)4);
+        timeUnitByteMap.put(TimeUnit.HOURS, (byte)5);
+        timeUnitByteMap.put(TimeUnit.DAYS, (byte)6);
+
+        for (TimeUnit value : TimeUnit.values()) {
+            assert timeUnitByteMap.containsKey(value) : value;
+        }
+
+        TIME_UNIT_BYTE_MAP = Collections.unmodifiableMap(timeUnitByteMap);
+    }
+
     private Version version = Version.CURRENT;
 
     /**
@@ -973,4 +995,24 @@ public abstract class StreamOutput extends OutputStream {
         writeVInt(enumValue.ordinal());
     }
 
+    /**
+     * Write a {@link TimeValue} to the stream
+     */
+    public void writeTimeValue(TimeValue timeValue) throws IOException {
+        writeZLong(timeValue.duration());
+        writeByte(TIME_UNIT_BYTE_MAP.get(timeValue.timeUnit()));
+    }
+
+    /**
+     * Write an optional {@link TimeValue} to the stream.
+     */
+    public void writeOptionalTimeValue(@Nullable TimeValue timeValue) throws IOException {
+        if (timeValue == null) {
+            writeBoolean(false);
+        } else {
+            writeBoolean(true);
+            writeTimeValue(timeValue);
+        }
+    }
+
 }

server/src/main/java/org/elasticsearch/common/unit/TimeValue.java

@@ -19,58 +19,19 @@
 
 package org.elasticsearch.common.unit;
 
-import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ToXContentFragment;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 
 import java.io.IOException;
-import java.util.Collections;
-import java.util.EnumMap;
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Locale;
-import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
-public class TimeValue implements Writeable, Comparable<TimeValue>, ToXContentFragment {
+public class TimeValue implements Comparable<TimeValue>, ToXContentFragment {
 
     /** How many nano-seconds in one milli-second */
     public static final long NSEC_PER_MSEC = TimeUnit.NANOSECONDS.convert(1, TimeUnit.MILLISECONDS);
 
-    private static Map<TimeUnit, Byte> TIME_UNIT_BYTE_MAP;
-    private static Map<Byte, TimeUnit> BYTE_TIME_UNIT_MAP;
-
-    static {
-        final Map<TimeUnit, Byte> timeUnitByteMap = new EnumMap<>(TimeUnit.class);
-        timeUnitByteMap.put(TimeUnit.NANOSECONDS, (byte)0);
-        timeUnitByteMap.put(TimeUnit.MICROSECONDS, (byte)1);
-        timeUnitByteMap.put(TimeUnit.MILLISECONDS, (byte)2);
-        timeUnitByteMap.put(TimeUnit.SECONDS, (byte)3);
-        timeUnitByteMap.put(TimeUnit.MINUTES, (byte)4);
-        timeUnitByteMap.put(TimeUnit.HOURS, (byte)5);
-        timeUnitByteMap.put(TimeUnit.DAYS, (byte)6);
-
-        final Set<Byte> bytes = new HashSet<>();
-        for (TimeUnit value : TimeUnit.values()) {
-            assert timeUnitByteMap.containsKey(value) : value;
-            assert bytes.add(timeUnitByteMap.get(value));
-        }
-
-        final Map<Byte, TimeUnit> byteTimeUnitMap = new HashMap<>();
-        for (Map.Entry<TimeUnit, Byte> entry : timeUnitByteMap.entrySet()) {
-            byteTimeUnitMap.put(entry.getValue(), entry.getKey());
-        }
-
-        TIME_UNIT_BYTE_MAP = Collections.unmodifiableMap(timeUnitByteMap);
-        BYTE_TIME_UNIT_MAP = Collections.unmodifiableMap(byteTimeUnitMap);
-    }
-
     public static final TimeValue MINUS_ONE = timeValueMillis(-1);
     public static final TimeValue ZERO = timeValueMillis(0);
 
@@ -96,15 +57,19 @@ public class TimeValue implements Writeable, Comparable<TimeValue>, ToXContentFr
 
     private final long duration;
 
-    // visible for testing
+    /**
-    long duration() {
+     * @return the number of {@link #timeUnit()} units this value contains
+     */
+    public long duration() {
         return duration;
     }
 
     private final TimeUnit timeUnit;
 
-    // visible for testing
+    /**
-    TimeUnit timeUnit() {
+     * @return the unit used for the this time value, see {@link #duration()}
+     */
+    public TimeUnit timeUnit() {
         return timeUnit;
     }
 
@@ -117,20 +82,6 @@ public class TimeValue implements Writeable, Comparable<TimeValue>, ToXContentFr
         this.timeUnit = timeUnit;
     }
 
-    /**
-     * Read from a stream.
-     */
-    public TimeValue(StreamInput in) throws IOException {
-        duration = in.readZLong();
-        timeUnit = BYTE_TIME_UNIT_MAP.get(in.readByte());
-    }
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        out.writeZLong(duration);
-        out.writeByte(TIME_UNIT_BYTE_MAP.get(timeUnit));
-    }
-
     public long nanos() {
         return timeUnit.toNanos(duration);
     }
@@ -271,7 +222,27 @@ public class TimeValue implements Writeable, Comparable<TimeValue>, ToXContentFr
             value = microsFrac();
             suffix = "micros";
         }
-        return Strings.format1Decimals(value, suffix);
+        return formatDecimal(value) + suffix;
+    }
+
+    private static String formatDecimal(double value) {
+        String p = String.valueOf(value);
+        int ix = p.indexOf('.') + 1;
+        int ex = p.indexOf('E');
+        char fraction = p.charAt(ix);
+        if (fraction == '0') {
+            if (ex != -1) {
+                return p.substring(0, ix - 1) + p.substring(ex);
+            } else {
+                return p.substring(0, ix - 1);
+            }
+        } else {
+            if (ex != -1) {
+                return p.substring(0, ix) + fraction + p.substring(ex);
+            } else {
+                return p.substring(0, ix) + fraction;
+            }
+        }
     }
 
     public String getStringRep() {
@@ -331,10 +302,8 @@ public class TimeValue implements Writeable, Comparable<TimeValue>, ToXContentFr
             return TimeValue.ZERO;
         } else {
             // Missing units:
-            throw new ElasticsearchParseException(
+            throw new IllegalArgumentException("failed to parse setting [" + settingName + "] with value [" + sValue +
-                "failed to parse setting [{}] with value [{}] as a time value: unit is missing or unrecognized",
+                    "] as a time value: unit is missing or unrecognized");
-                settingName,
-                sValue);
         }
     }
 
@@ -345,9 +314,9 @@ public class TimeValue implements Writeable, Comparable<TimeValue>, ToXContentFr
         } catch (final NumberFormatException e) {
             try {
                 @SuppressWarnings("unused") final double ignored = Double.parseDouble(s);
-                throw new ElasticsearchParseException("failed to parse [{}], fractional time values are not supported", e, initialInput);
+                throw new IllegalArgumentException("failed to parse [" + initialInput + "], fractional time values are not supported", e);
             } catch (final NumberFormatException ignored) {
-                throw new ElasticsearchParseException("failed to parse [{}]", e, initialInput);
+                throw new IllegalArgumentException("failed to parse [" + initialInput + "]", e);
             }
         }
     }

server/src/main/java/org/elasticsearch/discovery/zen/UnicastZenPing.java

@@ -627,7 +627,7 @@ public class UnicastZenPing extends AbstractComponent implements ZenPing {
         UnicastPingRequest(StreamInput in) throws IOException {
             super(in);
             id = in.readInt();
-            timeout = new TimeValue(in);
+            timeout = in.readTimeValue();
             pingResponse = new PingResponse(in);
         }
 
@@ -640,7 +640,7 @@ public class UnicastZenPing extends AbstractComponent implements ZenPing {
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeInt(id);
-            timeout.writeTo(out);
+            out.writeTimeValue(timeout);
             pingResponse.writeTo(out);
         }
     }

server/src/main/java/org/elasticsearch/index/reindex/AbstractBulkByScrollRequest.java

@@ -420,9 +420,9 @@ public abstract class AbstractBulkByScrollRequest<Self extends AbstractBulkByScr
         abortOnVersionConflict = in.readBoolean();
         size = in.readVInt();
         refresh = in.readBoolean();
-        timeout = new TimeValue(in);
+        timeout = in.readTimeValue();
         activeShardCount = ActiveShardCount.readFrom(in);
-        retryBackoffInitialTime = new TimeValue(in);
+        retryBackoffInitialTime = in.readTimeValue();
         maxRetries = in.readVInt();
         requestsPerSecond = in.readFloat();
         slices = in.readVInt();
@@ -435,9 +435,9 @@ public abstract class AbstractBulkByScrollRequest<Self extends AbstractBulkByScr
         out.writeBoolean(abortOnVersionConflict);
         out.writeVInt(size);
         out.writeBoolean(refresh);
-        timeout.writeTo(out);
+        out.writeTimeValue(timeout);
         activeShardCount.writeTo(out);
-        retryBackoffInitialTime.writeTo(out);
+        out.writeTimeValue(retryBackoffInitialTime);
         out.writeVInt(maxRetries);
         out.writeFloat(requestsPerSecond);
         if (out.getVersion().before(Version.V_6_1_0) && slices == AUTO_SLICES) {

server/src/main/java/org/elasticsearch/index/reindex/BulkByScrollResponse.java

@@ -152,7 +152,7 @@ public class BulkByScrollResponse extends ActionResponse implements ToXContentFr
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        took.writeTo(out);
+        out.writeTimeValue(took);
         status.writeTo(out);
         out.writeList(bulkFailures);
         out.writeList(searchFailures);
@@ -162,7 +162,7 @@ public class BulkByScrollResponse extends ActionResponse implements ToXContentFr
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        took = new TimeValue(in);
+        took = in.readTimeValue();
         status = new BulkByScrollTask.Status(in);
         bulkFailures = in.readList(Failure::new);
         searchFailures = in.readList(ScrollableHitSource.SearchFailure::new);

server/src/main/java/org/elasticsearch/index/reindex/BulkByScrollTask.java

@@ -317,10 +317,10 @@ public class BulkByScrollTask extends CancellableTask {
             noops = in.readVLong();
             bulkRetries = in.readVLong();
             searchRetries = in.readVLong();
-            throttled = new TimeValue(in);
+            throttled = in.readTimeValue();
             requestsPerSecond = in.readFloat();
             reasonCancelled = in.readOptionalString();
-            throttledUntil = new TimeValue(in);
+            throttledUntil = in.readTimeValue();
             if (in.getVersion().onOrAfter(Version.V_5_1_1)) {
                 sliceStatuses = in.readList(stream -> stream.readOptionalWriteable(StatusOrException::new));
             } else {
@@ -342,10 +342,10 @@ public class BulkByScrollTask extends CancellableTask {
             out.writeVLong(noops);
             out.writeVLong(bulkRetries);
             out.writeVLong(searchRetries);
-            throttled.writeTo(out);
+            out.writeTimeValue(throttled);
             out.writeFloat(requestsPerSecond);
             out.writeOptionalString(reasonCancelled);
-            throttledUntil.writeTo(out);
+            out.writeTimeValue(throttledUntil);
             if (out.getVersion().onOrAfter(Version.V_5_1_1)) {
                 out.writeVInt(sliceStatuses.size());
                 for (StatusOrException sliceStatus : sliceStatuses) {

server/src/main/java/org/elasticsearch/index/reindex/RemoteInfo.java

@@ -91,8 +91,8 @@ public class RemoteInfo implements Writeable {
         }
         this.headers = unmodifiableMap(headers);
         if (in.getVersion().onOrAfter(Version.V_5_2_0)) {
-            socketTimeout = new TimeValue(in);
+            socketTimeout = in.readTimeValue();
-            connectTimeout = new TimeValue(in);
+            connectTimeout = in.readTimeValue();
         } else {
             socketTimeout = DEFAULT_SOCKET_TIMEOUT;
             connectTimeout = DEFAULT_CONNECT_TIMEOUT;
@@ -113,8 +113,8 @@ public class RemoteInfo implements Writeable {
             out.writeString(header.getValue());
         }
         if (out.getVersion().onOrAfter(Version.V_5_2_0)) {
-            socketTimeout.writeTo(out);
+            out.writeTimeValue(socketTimeout);
-            connectTimeout.writeTo(out);
+            out.writeTimeValue(connectTimeout);
         }
     }
 

server/src/main/java/org/elasticsearch/search/Scroll.java

@@ -38,7 +38,7 @@ public final class Scroll implements Writeable {
     private final TimeValue keepAlive;
 
     public Scroll(StreamInput in) throws IOException {
-        this.keepAlive = new TimeValue(in);
+        this.keepAlive = in.readTimeValue();
     }
 
     /**
@@ -57,7 +57,7 @@ public final class Scroll implements Writeable {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        keepAlive.writeTo(out);
+        out.writeTimeValue(keepAlive);
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/builder/SearchSourceBuilder.java

@@ -224,7 +224,7 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
         }
         suggestBuilder = in.readOptionalWriteable(SuggestBuilder::new);
         terminateAfter = in.readVInt();
-        timeout = in.readOptionalWriteable(TimeValue::new);
+        timeout = in.readOptionalTimeValue();
         trackScores = in.readBoolean();
         version = in.readOptionalBoolean();
         extBuilders = in.readNamedWriteableList(SearchExtBuilder.class);
@@ -280,7 +280,7 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
         }
         out.writeOptionalWriteable(suggestBuilder);
         out.writeVInt(terminateAfter);
-        out.writeOptionalWriteable(timeout);
+        out.writeOptionalTimeValue(timeout);
         out.writeBoolean(trackScores);
         out.writeOptionalBoolean(version);
         out.writeNamedWriteableList(extBuilders);

server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

@@ -608,7 +608,7 @@ public class ThreadPool extends AbstractComponent implements Scheduler, Closeabl
             type = ThreadPoolType.fromType(in.readString());
             min = in.readInt();
             max = in.readInt();
-            keepAlive = in.readOptionalWriteable(TimeValue::new);
+            keepAlive = in.readOptionalTimeValue();
             queueSize = in.readOptionalWriteable(SizeValue::new);
         }
 
@@ -624,7 +624,7 @@ public class ThreadPool extends AbstractComponent implements Scheduler, Closeabl
             }
             out.writeInt(min);
             out.writeInt(max);
-            out.writeOptionalWriteable(keepAlive);
+            out.writeOptionalTimeValue(keepAlive);
             out.writeOptionalWriteable(queueSize);
         }
 

server/src/main/java/org/elasticsearch/transport/RemoteConnectionInfo.java

@@ -61,7 +61,7 @@ public final class RemoteConnectionInfo implements ToXContentFragment, Writeable
         seedNodes = input.readList(TransportAddress::new);
         httpAddresses = input.readList(TransportAddress::new);
         connectionsPerCluster = input.readVInt();
-        initialConnectionTimeout = new TimeValue(input);
+        initialConnectionTimeout = input.readTimeValue();
         numNodesConnected = input.readVInt();
         clusterAlias = input.readString();
         if (input.getVersion().onOrAfter(Version.V_6_1_0)) {
@@ -100,7 +100,7 @@ public final class RemoteConnectionInfo implements ToXContentFragment, Writeable
         out.writeList(seedNodes);
         out.writeList(httpAddresses);
         out.writeVInt(connectionsPerCluster);
-        initialConnectionTimeout.writeTo(out);
+        out.writeTimeValue(initialConnectionTimeout);
         out.writeVInt(numNodesConnected);
         out.writeString(clusterAlias);
         if (out.getVersion().onOrAfter(Version.V_6_1_0)) {

server/src/test/java/org/elasticsearch/common/unit/TimeValueTests.java

@@ -19,11 +19,9 @@
 
 package org.elasticsearch.common.unit;
 
-import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.test.ESTestCase;
-import org.joda.time.PeriodType;
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
@@ -108,10 +106,10 @@ public class TimeValueTests extends ESTestCase {
         // supported. Note that this is the only unit that is not case sensitive
         // as `m` is the only character that is overloaded in terms of which
         // time unit is expected between the upper and lower case versions
-        expectThrows(ElasticsearchParseException.class, () -> {
+        expectThrows(IllegalArgumentException.class, () -> {
             TimeValue.parseTimeValue("10 M", null, "test");
         });
-        expectThrows(ElasticsearchParseException.class, () -> {
+        expectThrows(IllegalArgumentException.class, () -> {
             TimeValue.parseTimeValue("10M", null, "test");
         });
 
@@ -132,8 +130,8 @@ public class TimeValueTests extends ESTestCase {
 
     public void testNonFractionalTimeValues() {
         final String s = randomAlphaOfLength(10) + randomTimeUnit();
-        final ElasticsearchParseException e =
+        final IllegalArgumentException e =
-            expectThrows(ElasticsearchParseException.class, () -> TimeValue.parseTimeValue(s, null, "test"));
+            expectThrows(IllegalArgumentException.class, () -> TimeValue.parseTimeValue(s, null, "test"));
         assertThat(e, hasToString(containsString("failed to parse [" + s + "]")));
         assertThat(e, not(hasToString(containsString(FRACTIONAL_TIME_VALUES_ARE_NOT_SUPPORTED))));
         assertThat(e.getCause(), instanceOf(NumberFormatException.class));
@@ -145,8 +143,8 @@ public class TimeValueTests extends ESTestCase {
             value = randomDouble();
         } while (value == 0);
         final String s = Double.toString(randomIntBetween(0, 128) + value) + randomTimeUnit();
-        final ElasticsearchParseException e =
+        final IllegalArgumentException e =
-            expectThrows(ElasticsearchParseException.class, () -> TimeValue.parseTimeValue(s, null, "test"));
+            expectThrows(IllegalArgumentException.class, () -> TimeValue.parseTimeValue(s, null, "test"));
         assertThat(e, hasToString(containsString("failed to parse [" + s + "]")));
         assertThat(e, hasToString(containsString(FRACTIONAL_TIME_VALUES_ARE_NOT_SUPPORTED)));
         assertThat(e.getCause(), instanceOf(NumberFormatException.class));
@@ -158,11 +156,11 @@ public class TimeValueTests extends ESTestCase {
 
     private void assertEqualityAfterSerialize(TimeValue value, int expectedSize) throws IOException {
         BytesStreamOutput out = new BytesStreamOutput();
-        value.writeTo(out);
+        out.writeTimeValue(value);
         assertEquals(expectedSize, out.size());
 
         StreamInput in = out.bytes().streamInput();
-        TimeValue inValue = new TimeValue(in);
+        TimeValue inValue = in.readTimeValue();
 
         assertThat(inValue, equalTo(value));
         assertThat(inValue.duration(), equalTo(value.duration()));
@@ -185,7 +183,7 @@ public class TimeValueTests extends ESTestCase {
         try {
             TimeValue.parseTimeValue("23tw", null, "test");
             fail("Expected ElasticsearchParseException");
-        } catch (ElasticsearchParseException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), containsString("failed to parse"));
         }
     }
@@ -194,7 +192,7 @@ public class TimeValueTests extends ESTestCase {
         try {
             TimeValue.parseTimeValue("42", null, "test");
             fail("Expected ElasticsearchParseException");
-        } catch (ElasticsearchParseException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), containsString("failed to parse"));
         }
     }
@@ -203,7 +201,7 @@ public class TimeValueTests extends ESTestCase {
         try {
             TimeValue.parseTimeValue("42ms.", null, "test");
             fail("Expected ElasticsearchParseException");
-        } catch (ElasticsearchParseException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), containsString("failed to parse"));
         }
     }

server/src/test/java/org/elasticsearch/script/ScriptServiceTests.java

@@ -116,8 +116,8 @@ public class ScriptServiceTests extends ESTestCase {
     public void testMaxCompilationRateSetting() throws Exception {
         assertThat(MAX_COMPILATION_RATE_FUNCTION.apply("10/1m"), is(Tuple.tuple(10, TimeValue.timeValueMinutes(1))));
         assertThat(MAX_COMPILATION_RATE_FUNCTION.apply("10/60s"), is(Tuple.tuple(10, TimeValue.timeValueMinutes(1))));
-        assertException("10/m", ElasticsearchParseException.class, "failed to parse [m]");
+        assertException("10/m", IllegalArgumentException.class, "failed to parse [m]");
-        assertException("6/1.6m", ElasticsearchParseException.class, "failed to parse [1.6m], fractional time values are not supported");
+        assertException("6/1.6m", IllegalArgumentException.class, "failed to parse [1.6m], fractional time values are not supported");
         assertException("foo/bar", IllegalArgumentException.class, "could not parse [foo] as integer in value [foo/bar]");
         assertException("6.0/1m", IllegalArgumentException.class, "could not parse [6.0] as integer in value [6.0/1m]");
         assertException("6/-1m", IllegalArgumentException.class, "time value [-1m] must be positive");

server/src/test/java/org/elasticsearch/search/builder/SearchSourceBuilderTests.java

@@ -345,7 +345,7 @@ public class SearchSourceBuilderTests extends AbstractSearchTestCase {
         final int timeout = randomIntBetween(1, 1024);
         final String query = "{ \"query\": { \"match_all\": {}}, \"timeout\": \"" + timeout + "\"}";
         try (XContentParser parser = createParser(JsonXContent.jsonXContent, query)) {
-            final ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> SearchSourceBuilder.fromXContent(
+            final IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> SearchSourceBuilder.fromXContent(
                     parser));
             assertThat(e, hasToString(containsString("unit is missing or unrecognized")));
         }