From f051103067a1dc0d86121adfbfe7e527c94795cc Mon Sep 17 00:00:00 2001 From: Nicholas Walter Knize Date: Mon, 8 May 2023 12:52:44 -0500 Subject: [PATCH 1/3] [Refactor] Stream Reader and Write Generics StreamInput and StreamOutput provide the core functionality for marshalling / unmarshalling objects over the transport wire. The class is intended to be generic but it is tightly coupled to the types defined in the server module. Because of this tight coupling, the classes cannot be refactored to the core library, thus all new types are required to be hard coded in the server module. To decouple this logic and make it more generic across opensearch modules and plugins, this commit introduces a reader and writer registry in a new BaseWriteable interface. The StreamInput and StreamOutput also now inherits from new BaseStreamInput and BaseStreamOutput classes, respectively, located in the core library. This will be the new home for streaming primitives in a follow up commit. Signed-off-by: Nicholas Walter Knize --- libs/core/build.gradle | 5 + .../common/io/stream/BaseStreamInput.java | 19 +++ .../common/io/stream/BaseStreamOutput.java | 19 +++ .../core/common/io/stream/BaseWriteable.java | 122 ++++++++++++++++++ .../core/common/io/stream/package-info.java | 9 ++ .../bucket/geogrid/BaseGeoGrid.java | 5 +- .../geogrid/InternalGeoHashGridBucket.java | 5 +- .../geogrid/InternalGeoTileGridBucket.java | 5 +- .../opensearch/common/geo/GeoBoundingBox.java | 8 +- .../org/opensearch/common/geo/GeoPoint.java | 26 ++++ .../common/io/stream/StreamInput.java | 32 ++--- .../common/io/stream/StreamOutput.java | 39 ++---- .../common/io/stream/Writeable.java | 27 +--- .../index/query/GeoDistanceQueryBuilder.java | 4 +- .../index/query/GeoPolygonQueryBuilder.java | 4 +- .../script/JodaCompatibleZonedDateTime.java | 27 ++++ .../transport/TransportService.java | 19 +++ .../common/io/stream/BaseStreamTests.java | 14 ++ .../common/io/stream/BytesStreamsTests.java | 4 +- .../opensearch/test/OpenSearchTestCase.java | 2 + 20 files changed, 305 insertions(+), 90 deletions(-) create mode 100644 libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.java create mode 100644 libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.java create mode 100644 libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java create mode 100644 libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.java diff --git a/libs/core/build.gradle b/libs/core/build.gradle index 696ea11f5be50..e24b277470f69 100644 --- a/libs/core/build.gradle +++ b/libs/core/build.gradle @@ -117,3 +117,8 @@ tasks.named("dependencyLicenses").configure { mapping from: /jackson-.*/, to: 'jackson' mapping from: /lucene-.*/, to: 'lucene' } + +tasks.withType(JavaCompile).configureEach { + options.compilerArgs -= '-Xlint:rawtypes' + options.compilerArgs -= '-Xlint:unchecked' +} diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.java new file mode 100644 index 0000000000000..a0e50722bf01d --- /dev/null +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.java @@ -0,0 +1,19 @@ +/* + * 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.core.common.io.stream; + +import java.io.InputStream; + +/** + * Foundation class for reading core types off the transport stream + * + * todo: refactor {@code StreamInput} primitive readers to this class + * + * @opensearch.internal + */ +public abstract class BaseStreamInput extends InputStream {} diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.java new file mode 100644 index 0000000000000..f7a8862fa5f2c --- /dev/null +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.java @@ -0,0 +1,19 @@ +/* + * 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.core.common.io.stream; + +import java.io.OutputStream; + +/** + * Foundation class for writing core types over the transport stream + * + * todo: refactor {@code StreamOutput} primitive writers to this class + * + * @opensearch.internal + */ +public abstract class BaseStreamOutput extends OutputStream {} diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java new file mode 100644 index 0000000000000..8d68b010acf46 --- /dev/null +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java @@ -0,0 +1,122 @@ +/* + * 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.core.common.io.stream; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +/** + * Implementers can be written to a {@code StreamOutput} and read from a {@code StreamInput}. This allows them to be "thrown + * across the wire" using OpenSearch's internal protocol. If the implementer also implements equals and hashCode then a copy made by + * serializing and deserializing must be equal and have the same hashCode. It isn't required that such a copy be entirely unchanged. + * + * @opensearch.internal + */ +public interface BaseWriteable { + /** + * A WriteableRegistry registers {@link Writer} methods for writing data types over a + * {@link BaseStreamOutput} channel and {@link Reader} methods for reading data from a + * {@link BaseStreamInput} channel. + * + * @opensearch.internal + */ + class WriteableRegistry { + private static final Map, Writer> WRITER_REGISTRY = new HashMap<>(); + private static final Map READER_REGISTRY = new HashMap<>(); + + /** + * registers a streamable writer + * + * @opensearch.internal + */ + public static void registerWriter(final Class clazz, final W writer) { + if (WRITER_REGISTRY.containsKey(clazz)) { + throw new IllegalArgumentException("Streamable writer already registered for type [" + clazz.getName() + "]"); + } + WRITER_REGISTRY.put(clazz, writer); + } + + /** + * registers a streamable reader + * + * @opensearch.internal + */ + public static void registerReader(final byte ordinal, final R reader) { + if (READER_REGISTRY.containsKey(ordinal)) { + throw new IllegalArgumentException("Streamable reader already registered for ordinal [" + (int) ordinal + "]"); + } + READER_REGISTRY.put(ordinal, reader); + } + + public static W getWriter(final Class clazz) { + return (W) WRITER_REGISTRY.get(clazz); + } + + public static R getReader(final byte b) { + return (R) READER_REGISTRY.get(b); + } + } + + /** + * Write this into the {@linkplain BaseStreamOutput}. + */ + void writeTo(final S out) throws IOException; + + /** + * Reference to a method that can write some object to a {@link BaseStreamOutput}. + *

+ * By convention this is a method from {@link BaseStreamOutput} itself (e.g., {@code StreamOutput#writeString}). If the value can be + * {@code null}, then the "optional" variant of methods should be used! + *

+ * Most classes should implement {@code Writeable} and the {@code Writeable#writeTo(BaseStreamOutput)} method should use + * {@link BaseStreamOutput} methods directly or this indirectly: + * 


+     * public void writeTo(StreamOutput out) throws IOException {
+     *     out.writeVInt(someValue);
+     *     out.writeMapOfLists(someMap, StreamOutput::writeString, StreamOutput::writeString);
+     * }
+     *
+ */ + @FunctionalInterface + interface Writer { + + /** + * Write {@code V}-type {@code value} to the {@code out}put stream. + * + * @param out Output to write the {@code value} too + * @param value The value to add + */ + void write(final S out, V value) throws IOException; + } + + /** + * Reference to a method that can read some object from a stream. By convention this is a constructor that takes + * {@linkplain BaseStreamInput} as an argument for most classes and a static method for things like enums. Returning null from one of these + * is always wrong - for that we use methods like {@code StreamInput#readOptionalWriteable(Reader)}. + *

+ * As most classes will implement this via a constructor (or a static method in the case of enumerations), it's something that should + * look like: + * 


+     * public MyClass(final StreamInput in) throws IOException {
+     *     this.someValue = in.readVInt();
+     *     this.someMap = in.readMapOfLists(StreamInput::readString, StreamInput::readString);
+     * }
+     *
+ */ + @FunctionalInterface + interface Reader { + + /** + * Read {@code V}-type value from a stream. + * + * @param in Input to read the value from + */ + V read(final S in) throws IOException; + } +} diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.java new file mode 100644 index 0000000000000..76d0842466b96 --- /dev/null +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.java @@ -0,0 +1,9 @@ +/* + * 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. + */ +/** Core transport stream classes */ +package org.opensearch.core.common.io.stream; diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/BaseGeoGrid.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/BaseGeoGrid.java index f8cfcabc10593..347b49cdf9981 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/BaseGeoGrid.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/BaseGeoGrid.java @@ -35,6 +35,7 @@ import org.opensearch.common.io.stream.StreamOutput; import org.opensearch.common.io.stream.Writeable; import org.opensearch.common.util.LongObjectPagedHashMap; +import org.opensearch.core.common.io.stream.BaseWriteable; import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.search.aggregations.InternalAggregation; import org.opensearch.search.aggregations.InternalAggregations; @@ -69,7 +70,7 @@ protected BaseGeoGrid(String name, int requiredSize, List buc this.buckets = buckets; } - protected abstract Writeable.Reader getBucketReader(); + protected abstract BaseWriteable.Reader getBucketReader(); /** * Read from a stream. @@ -77,7 +78,7 @@ protected BaseGeoGrid(String name, int requiredSize, List buc public BaseGeoGrid(StreamInput in) throws IOException { super(in); requiredSize = readSize(in); - buckets = (List) in.readList(getBucketReader()); + buckets = (List) in.readList((Writeable.Reader) getBucketReader()); } @Override diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGridBucket.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGridBucket.java index 6e7ed8a679681..493c84ed9dfd7 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGridBucket.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGridBucket.java @@ -33,6 +33,7 @@ import org.opensearch.common.geo.GeoPoint; import org.opensearch.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.BaseStreamInput; import org.opensearch.geometry.utils.Geohash; import org.opensearch.search.aggregations.InternalAggregations; @@ -51,8 +52,8 @@ class InternalGeoHashGridBucket extends BaseGeoGridBucket>registerWriter(GeoPoint.class, (o, v) -> { + o.writeByte((byte) 22); + ((GeoPoint) v).writeTo(o); + }); + WriteableRegistry.>registerReader(Byte.valueOf((byte) 22), GeoPoint::new); + } + public GeoPoint reset(double lat, double lon) { this.lat = lat; this.lon = lon; @@ -210,6 +231,11 @@ public GeoPoint resetFromGeoHash(long geohashLong) { return this.resetFromIndexHash(BitUtil.flipFlop((geohashLong >>> 4) << ((level * 5) + 2))); } + public void writeTo(final StreamOutput out) throws IOException { + out.writeDouble(this.lat); + out.writeDouble(this.lon); + } + public double lat() { return this.lat; } diff --git a/server/src/main/java/org/opensearch/common/io/stream/StreamInput.java b/server/src/main/java/org/opensearch/common/io/stream/StreamInput.java index f1b4ffe2219aa..65428c8114b43 100644 --- a/server/src/main/java/org/opensearch/common/io/stream/StreamInput.java +++ b/server/src/main/java/org/opensearch/common/io/stream/StreamInput.java @@ -50,20 +50,18 @@ import org.opensearch.common.Strings; import org.opensearch.common.bytes.BytesArray; import org.opensearch.common.bytes.BytesReference; -import org.opensearch.common.geo.GeoPoint; import org.opensearch.common.settings.SecureString; import org.opensearch.common.text.Text; -import org.opensearch.common.time.DateUtils; import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.common.io.stream.BaseStreamInput; +import org.opensearch.core.common.io.stream.BaseWriteable; import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException; -import org.opensearch.script.JodaCompatibleZonedDateTime; import java.io.ByteArrayInputStream; import java.io.EOFException; import java.io.FileNotFoundException; import java.io.FilterInputStream; import java.io.IOException; -import java.io.InputStream; import java.math.BigInteger; import java.nio.file.AccessDeniedException; import java.nio.file.AtomicMoveNotSupportedException; @@ -108,7 +106,7 @@ * * @opensearch.internal */ -public abstract class StreamInput extends InputStream { +public abstract class StreamInput extends BaseStreamInput { private Version version = Version.CURRENT; @@ -686,6 +684,11 @@ public Map readMap() throws IOException { @Nullable public Object readGenericValue() throws IOException { byte type = readByte(); + BaseWriteable.Reader r = BaseWriteable.WriteableRegistry.getReader(type); + if (r != null) { + return r.read(this); + } + switch (type) { case -1: return null; @@ -715,8 +718,6 @@ public Object readGenericValue() throws IOException { return readByte(); case 12: return readDate(); - case 13: - return readDateTime(); case 14: return readBytesReference(); case 15: @@ -733,8 +734,6 @@ public Object readGenericValue() throws IOException { return readDoubleArray(); case 21: return readBytesRef(); - case 22: - return readGeoPoint(); case 23: return readZonedDateTime(); case 24: @@ -778,14 +777,6 @@ private List readArrayList() throws IOException { return list; } - private JodaCompatibleZonedDateTime readDateTime() throws IOException { - // we reuse DateTime to communicate with older nodes that don't know about the joda compat layer, but - // here we are on a new node so we always want a compat datetime - final ZoneId zoneId = DateUtils.dateTimeZoneToZoneId(DateTimeZone.forID(readString())); - long millis = readLong(); - return new JodaCompatibleZonedDateTime(Instant.ofEpochMilli(millis), zoneId); - } - private ZonedDateTime readZonedDateTime() throws IOException { final String timeZoneId = readString(); return ZonedDateTime.ofInstant(Instant.ofEpochMilli(readLong()), ZoneId.of(timeZoneId)); @@ -833,13 +824,6 @@ private Date readDate() throws IOException { return new Date(readLong()); } - /** - * Reads a {@link GeoPoint} from this stream input - */ - public GeoPoint readGeoPoint() throws IOException { - return new GeoPoint(readDouble(), readDouble()); - } - /** * Read a {@linkplain DateTimeZone}. */ diff --git a/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java b/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java index da17cddfcf97b..3df6e9752c8b2 100644 --- a/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java +++ b/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java @@ -51,18 +51,17 @@ import org.opensearch.common.Nullable; import org.opensearch.common.bytes.BytesArray; import org.opensearch.common.bytes.BytesReference; -import org.opensearch.common.geo.GeoPoint; import org.opensearch.common.io.stream.Writeable.Writer; import org.opensearch.common.settings.SecureString; import org.opensearch.common.text.Text; import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.common.io.stream.BaseStreamOutput; +import org.opensearch.core.common.io.stream.BaseWriteable; import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException; -import org.opensearch.script.JodaCompatibleZonedDateTime; import java.io.EOFException; import java.io.FileNotFoundException; import java.io.IOException; -import java.io.OutputStream; import java.math.BigInteger; import java.nio.file.AccessDeniedException; import java.nio.file.AtomicMoveNotSupportedException; @@ -103,7 +102,7 @@ * * @opensearch.internal */ -public abstract class StreamOutput extends OutputStream { +public abstract class StreamOutput extends BaseStreamOutput { private static final int MAX_NESTED_EXCEPTION_LEVEL = 100; @@ -663,10 +662,10 @@ public final void writeOptionalInstant(@Nullable Instant instant) throws IOExcep } } - private static final Map, Writer> WRITERS; + private static final Map, BaseWriteable.Writer> WRITERS; static { - Map, Writer> writers = new HashMap<>(); + Map, BaseWriteable.Writer> writers = new HashMap<>(); writers.put(String.class, (o, v) -> { o.writeByte((byte) 0); o.writeString((String) v); @@ -773,25 +772,12 @@ public final void writeOptionalInstant(@Nullable Instant instant) throws IOExcep o.writeByte((byte) 21); o.writeBytesRef((BytesRef) v); }); - writers.put(GeoPoint.class, (o, v) -> { - o.writeByte((byte) 22); - o.writeGeoPoint((GeoPoint) v); - }); writers.put(ZonedDateTime.class, (o, v) -> { o.writeByte((byte) 23); final ZonedDateTime zonedDateTime = (ZonedDateTime) v; o.writeString(zonedDateTime.getZone().getId()); o.writeLong(zonedDateTime.toInstant().toEpochMilli()); }); - writers.put(JodaCompatibleZonedDateTime.class, (o, v) -> { - // write the joda compatibility datetime as joda datetime - o.writeByte((byte) 13); - final JodaCompatibleZonedDateTime zonedDateTime = (JodaCompatibleZonedDateTime) v; - String zoneId = zonedDateTime.getZonedDateTime().getZone().getId(); - // joda does not understand "Z" for utc, so we must special case - o.writeString(zoneId.equals("Z") ? DateTimeZone.UTC.getID() : zoneId); - o.writeLong(zonedDateTime.toInstant().toEpochMilli()); - }); writers.put(Set.class, (o, v) -> { if (v instanceof LinkedHashSet) { o.writeByte((byte) 24); @@ -838,7 +824,12 @@ public void writeGenericValue(@Nullable Object value) throws IOException { return; } final Class type = getGenericType(value); - final Writer writer = WRITERS.get(type); + BaseWriteable.Writer writer = BaseWriteable.WriteableRegistry.getWriter(type); + if (writer == null) { + // fallback to this local hashmap + // todo: move all writers to the registry + writer = WRITERS.get(type); + } if (writer != null) { writer.write(this, value); } else { @@ -1145,14 +1136,6 @@ public void writeOptionalNamedWriteable(@Nullable NamedWriteable namedWriteable) } } - /** - * Writes the given {@link GeoPoint} to the stream - */ - public void writeGeoPoint(GeoPoint geoPoint) throws IOException { - writeDouble(geoPoint.lat()); - writeDouble(geoPoint.lon()); - } - /** * Write a {@linkplain DateTimeZone} to the stream. */ diff --git a/server/src/main/java/org/opensearch/common/io/stream/Writeable.java b/server/src/main/java/org/opensearch/common/io/stream/Writeable.java index 5fd227db6ca83..c04cd7977fdc0 100644 --- a/server/src/main/java/org/opensearch/common/io/stream/Writeable.java +++ b/server/src/main/java/org/opensearch/common/io/stream/Writeable.java @@ -32,6 +32,8 @@ package org.opensearch.common.io.stream; +import org.opensearch.core.common.io.stream.BaseWriteable; + import java.io.IOException; /** @@ -41,7 +43,7 @@ * * @opensearch.internal */ -public interface Writeable { +public interface Writeable extends BaseWriteable { /** * Write this into the {@linkplain StreamOutput}. @@ -64,17 +66,7 @@ public interface Writeable { * */ @FunctionalInterface - interface Writer { - - /** - * Write {@code V}-type {@code value} to the {@code out}put stream. - * - * @param out Output to write the {@code value} too - * @param value The value to add - */ - void write(StreamOutput out, V value) throws IOException; - - } + interface Writer extends BaseWriteable.Writer {} /** * Reference to a method that can read some object from a stream. By convention this is a constructor that takes @@ -91,15 +83,6 @@ interface Writer { * */ @FunctionalInterface - interface Reader { - - /** - * Read {@code V}-type value from a stream. - * - * @param in Input to read the value from - */ - V read(StreamInput in) throws IOException; - - } + interface Reader extends BaseWriteable.Reader {} } diff --git a/server/src/main/java/org/opensearch/index/query/GeoDistanceQueryBuilder.java b/server/src/main/java/org/opensearch/index/query/GeoDistanceQueryBuilder.java index e3ad2781f5546..bc52d8fe6a6df 100644 --- a/server/src/main/java/org/opensearch/index/query/GeoDistanceQueryBuilder.java +++ b/server/src/main/java/org/opensearch/index/query/GeoDistanceQueryBuilder.java @@ -113,7 +113,7 @@ public GeoDistanceQueryBuilder(StreamInput in) throws IOException { fieldName = in.readString(); distance = in.readDouble(); validationMethod = GeoValidationMethod.readFromStream(in); - center = in.readGeoPoint(); + center = new GeoPoint(in); geoDistance = GeoDistance.readFromStream(in); ignoreUnmapped = in.readBoolean(); } @@ -123,7 +123,7 @@ protected void doWriteTo(StreamOutput out) throws IOException { out.writeString(fieldName); out.writeDouble(distance); validationMethod.writeTo(out); - out.writeGeoPoint(center); + center.writeTo(out); geoDistance.writeTo(out); out.writeBoolean(ignoreUnmapped); } diff --git a/server/src/main/java/org/opensearch/index/query/GeoPolygonQueryBuilder.java b/server/src/main/java/org/opensearch/index/query/GeoPolygonQueryBuilder.java index e0c7d44a08fac..f5e40fec78c25 100644 --- a/server/src/main/java/org/opensearch/index/query/GeoPolygonQueryBuilder.java +++ b/server/src/main/java/org/opensearch/index/query/GeoPolygonQueryBuilder.java @@ -114,7 +114,7 @@ public GeoPolygonQueryBuilder(StreamInput in) throws IOException { int size = in.readVInt(); shell = new ArrayList<>(size); for (int i = 0; i < size; i++) { - shell.add(in.readGeoPoint()); + shell.add(new GeoPoint(in)); } validationMethod = GeoValidationMethod.readFromStream(in); ignoreUnmapped = in.readBoolean(); @@ -125,7 +125,7 @@ protected void doWriteTo(StreamOutput out) throws IOException { out.writeString(fieldName); out.writeVInt(shell.size()); for (GeoPoint point : shell) { - out.writeGeoPoint(point); + point.writeTo(out); } validationMethod.writeTo(out); out.writeBoolean(ignoreUnmapped); diff --git a/server/src/main/java/org/opensearch/script/JodaCompatibleZonedDateTime.java b/server/src/main/java/org/opensearch/script/JodaCompatibleZonedDateTime.java index 08306b3f275a8..8d14a4fae992d 100644 --- a/server/src/main/java/org/opensearch/script/JodaCompatibleZonedDateTime.java +++ b/server/src/main/java/org/opensearch/script/JodaCompatibleZonedDateTime.java @@ -32,8 +32,15 @@ package org.opensearch.script; +import org.joda.time.DateTimeZone; import org.opensearch.common.SuppressForbidden; +import org.opensearch.common.io.stream.StreamInput; +import org.opensearch.common.io.stream.StreamOutput; +import org.opensearch.core.common.io.stream.BaseWriteable.Reader; +import org.opensearch.core.common.io.stream.BaseWriteable.Writer; import org.opensearch.common.time.DateFormatter; +import org.opensearch.common.time.DateUtils; +import org.opensearch.core.common.io.stream.BaseWriteable.WriteableRegistry; import java.time.DayOfWeek; import java.time.Instant; @@ -77,6 +84,26 @@ public JodaCompatibleZonedDateTime(Instant instant, ZoneId zone) { this.dt = ZonedDateTime.ofInstant(instant, zone); } + /** + * Register this type as a streamable so it can be serialized over the wire + */ + public static void registerStreamables() { + WriteableRegistry.>registerWriter(JodaCompatibleZonedDateTime.class, (o, v) -> { + // write the joda compatibility datetime as joda datetime + o.writeByte((byte) 13); + final JodaCompatibleZonedDateTime zonedDateTime = (JodaCompatibleZonedDateTime) v; + String zoneId = zonedDateTime.getZonedDateTime().getZone().getId(); + // joda does not understand "Z" for utc, so we must special case + o.writeString(zoneId.equals("Z") ? DateTimeZone.UTC.getID() : zoneId); + o.writeLong(zonedDateTime.toInstant().toEpochMilli()); + }); + WriteableRegistry.>registerReader(Byte.valueOf((byte) 13), (i) -> { + final ZoneId zoneId = DateUtils.dateTimeZoneToZoneId(DateTimeZone.forID(i.readString())); + long millis = i.readLong(); + return new JodaCompatibleZonedDateTime(Instant.ofEpochMilli(millis), zoneId); + }); + } + // access the underlying ZonedDateTime public ZonedDateTime getZonedDateTime() { return dt; diff --git a/server/src/main/java/org/opensearch/transport/TransportService.java b/server/src/main/java/org/opensearch/transport/TransportService.java index 2b5dee229b8cb..e96f1d11c89d0 100644 --- a/server/src/main/java/org/opensearch/transport/TransportService.java +++ b/server/src/main/java/org/opensearch/transport/TransportService.java @@ -44,6 +44,7 @@ import org.opensearch.common.Nullable; import org.opensearch.common.Strings; import org.opensearch.common.component.AbstractLifecycleComponent; +import org.opensearch.common.geo.GeoPoint; import org.opensearch.common.io.stream.StreamInput; import org.opensearch.common.io.stream.StreamOutput; import org.opensearch.common.io.stream.Writeable; @@ -61,6 +62,7 @@ import org.opensearch.common.util.io.IOUtils; import org.opensearch.node.NodeClosedException; import org.opensearch.node.ReportingService; +import org.opensearch.script.JodaCompatibleZonedDateTime; import org.opensearch.tasks.Task; import org.opensearch.tasks.TaskManager; import org.opensearch.threadpool.Scheduler; @@ -161,6 +163,14 @@ public boolean isClosed() { public void close() {} }; + static { + // registers server specific streamables + registerStreamables(); + } + + /** does nothing. easy way to ensure class is loaded */ + public static void ensureClassloaded() {} + /** * Build the service. * @@ -231,6 +241,15 @@ public TransportService( ); } + /** + * Registers server specific types as a streamables for serialization + * over the {@link StreamOutput} and {@link StreamInput} wire + */ + private static void registerStreamables() { + JodaCompatibleZonedDateTime.registerStreamables(); + GeoPoint.registerStreamables(); + } + public RemoteClusterService getRemoteClusterService() { return remoteClusterService; } diff --git a/server/src/test/java/org/opensearch/common/io/stream/BaseStreamTests.java b/server/src/test/java/org/opensearch/common/io/stream/BaseStreamTests.java index b92e59e43e0db..bd970be5e977d 100644 --- a/server/src/test/java/org/opensearch/common/io/stream/BaseStreamTests.java +++ b/server/src/test/java/org/opensearch/common/io/stream/BaseStreamTests.java @@ -40,12 +40,14 @@ import org.opensearch.common.bytes.BytesReference; import org.opensearch.common.collect.Tuple; import org.opensearch.common.settings.SecureString; +import org.opensearch.script.JodaCompatibleZonedDateTime; import org.opensearch.test.OpenSearchTestCase; import java.io.ByteArrayInputStream; import java.io.EOFException; import java.io.IOException; import java.time.Instant; +import java.time.ZoneOffset; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -396,6 +398,18 @@ public void testOptionalInstantSerialization() throws IOException { } } + public void testJodaDateTimeSerialization() throws IOException { + final BytesStreamOutput output = new BytesStreamOutput(); + long millis = randomIntBetween(0, Integer.MAX_VALUE); + JodaCompatibleZonedDateTime time = new JodaCompatibleZonedDateTime(Instant.ofEpochMilli(millis), ZoneOffset.ofHours(-7)); + output.writeGenericValue(time); + + final BytesReference bytesReference = output.bytes(); + final StreamInput input = getStreamInput(bytesReference); + Object inTime = input.readGenericValue(); + assertEquals(time, inTime); + } + static final class WriteableString implements Writeable { final String string; diff --git a/server/src/test/java/org/opensearch/common/io/stream/BytesStreamsTests.java b/server/src/test/java/org/opensearch/common/io/stream/BytesStreamsTests.java index bff35ad9fc975..36b84560bce41 100644 --- a/server/src/test/java/org/opensearch/common/io/stream/BytesStreamsTests.java +++ b/server/src/test/java/org/opensearch/common/io/stream/BytesStreamsTests.java @@ -639,9 +639,9 @@ public void testReadWriteGeoPoint() throws IOException { try (BytesStreamOutput out = new BytesStreamOutput()) { GeoPoint geoPoint = new GeoPoint(randomDouble(), randomDouble()); - out.writeGeoPoint(geoPoint); + geoPoint.writeTo(out); StreamInput wrap = out.bytes().streamInput(); - GeoPoint point = wrap.readGeoPoint(); + GeoPoint point = new GeoPoint(wrap); assertEquals(point, geoPoint); } } diff --git a/test/framework/src/main/java/org/opensearch/test/OpenSearchTestCase.java b/test/framework/src/main/java/org/opensearch/test/OpenSearchTestCase.java index aed9064c9b09f..ca416cafe8ea4 100644 --- a/test/framework/src/main/java/org/opensearch/test/OpenSearchTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/OpenSearchTestCase.java @@ -129,6 +129,7 @@ import org.opensearch.test.junit.listeners.LoggingListener; import org.opensearch.test.junit.listeners.ReproduceInfoPrinter; import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; import org.opensearch.transport.nio.MockNioTransportPlugin; import org.joda.time.DateTimeZone; import org.junit.After; @@ -258,6 +259,7 @@ public void append(LogEvent event) { })); BootstrapForTesting.ensureInitialized(); + TransportService.ensureClassloaded(); // ensure server streamables are registered // filter out joda timezones that are deprecated for the java time migration List jodaTZIds = DateTimeZone.getAvailableIDs() From 66f3a9691c589f877261d7b6ded7dbb876e42446 Mon Sep 17 00:00:00 2001 From: Nicholas Walter Knize Date: Wed, 10 May 2023 08:56:59 -0500 Subject: [PATCH 2/3] Remove unchecked and raw warning suppression Signed-off-by: Nicholas Walter Knize --- libs/core/build.gradle | 5 ----- .../core/common/io/stream/BaseWriteable.java | 22 +++++++++++++------ 2 files changed, 15 insertions(+), 12 deletions(-) diff --git a/libs/core/build.gradle b/libs/core/build.gradle index e24b277470f69..696ea11f5be50 100644 --- a/libs/core/build.gradle +++ b/libs/core/build.gradle @@ -117,8 +117,3 @@ tasks.named("dependencyLicenses").configure { mapping from: /jackson-.*/, to: 'jackson' mapping from: /lucene-.*/, to: 'lucene' } - -tasks.withType(JavaCompile).configureEach { - options.compilerArgs -= '-Xlint:rawtypes' - options.compilerArgs -= '-Xlint:unchecked' -} diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java index 8d68b010acf46..56172e7c6a50e 100644 --- a/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java @@ -8,8 +8,8 @@ package org.opensearch.core.common.io.stream; import java.io.IOException; -import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; /** * Implementers can be written to a {@code StreamOutput} and read from a {@code StreamInput}. This allows them to be "thrown @@ -27,15 +27,15 @@ public interface BaseWriteable { * @opensearch.internal */ class WriteableRegistry { - private static final Map, Writer> WRITER_REGISTRY = new HashMap<>(); - private static final Map READER_REGISTRY = new HashMap<>(); + private static final Map, Writer> WRITER_REGISTRY = new ConcurrentHashMap<>(); + private static final Map> READER_REGISTRY = new ConcurrentHashMap<>(); /** * registers a streamable writer * * @opensearch.internal */ - public static void registerWriter(final Class clazz, final W writer) { + public static > void registerWriter(final Class clazz, final W writer) { if (WRITER_REGISTRY.containsKey(clazz)) { throw new IllegalArgumentException("Streamable writer already registered for type [" + clazz.getName() + "]"); } @@ -47,18 +47,26 @@ public static void registerWriter(final Class clazz, final * * @opensearch.internal */ - public static void registerReader(final byte ordinal, final R reader) { + public static > void registerReader(final byte ordinal, final R reader) { if (READER_REGISTRY.containsKey(ordinal)) { throw new IllegalArgumentException("Streamable reader already registered for ordinal [" + (int) ordinal + "]"); } READER_REGISTRY.put(ordinal, reader); } - public static W getWriter(final Class clazz) { + /** + * Returns the registered writer keyed by the class type + */ + @SuppressWarnings("unchecked") + public static > W getWriter(final Class clazz) { return (W) WRITER_REGISTRY.get(clazz); } - public static R getReader(final byte b) { + /** + * Returns the ristered reader keyed by the unique ordinal + */ + @SuppressWarnings("unchecked") + public static > R getReader(final byte b) { return (R) READER_REGISTRY.get(b); } } From f71b607904c01dabede64914f7d17b38c89f3a7c Mon Sep 17 00:00:00 2001 From: Nicholas Walter Knize Date: Thu, 11 May 2023 13:16:44 -0500 Subject: [PATCH 3/3] cleanup PR Signed-off-by: Nicholas Walter Knize --- .../search/aggregations/bucket/geogrid/BaseGeoGrid.java | 6 ++---- .../search/aggregations/bucket/geogrid/GeoHashGrid.java | 2 +- .../search/aggregations/bucket/geogrid/GeoTileGrid.java | 2 +- .../bucket/geogrid/InternalGeoHashGridBucket.java | 5 ++--- .../bucket/geogrid/InternalGeoTileGridBucket.java | 5 ++--- .../org/opensearch/common/io/stream/StreamInput.java | 9 ++++++--- .../org/opensearch/common/io/stream/StreamOutput.java | 6 +++--- 7 files changed, 17 insertions(+), 18 deletions(-) diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/BaseGeoGrid.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/BaseGeoGrid.java index 347b49cdf9981..9b5ed7777204e 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/BaseGeoGrid.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/BaseGeoGrid.java @@ -33,9 +33,7 @@ import org.opensearch.common.io.stream.StreamInput; import org.opensearch.common.io.stream.StreamOutput; -import org.opensearch.common.io.stream.Writeable; import org.opensearch.common.util.LongObjectPagedHashMap; -import org.opensearch.core.common.io.stream.BaseWriteable; import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.search.aggregations.InternalAggregation; import org.opensearch.search.aggregations.InternalAggregations; @@ -70,7 +68,7 @@ protected BaseGeoGrid(String name, int requiredSize, List buc this.buckets = buckets; } - protected abstract BaseWriteable.Reader getBucketReader(); + protected abstract Reader getBucketReader(); /** * Read from a stream. @@ -78,7 +76,7 @@ protected BaseGeoGrid(String name, int requiredSize, List buc public BaseGeoGrid(StreamInput in) throws IOException { super(in); requiredSize = readSize(in); - buckets = (List) in.readList((Writeable.Reader) getBucketReader()); + buckets = (List) in.readList(getBucketReader()); } @Override diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/GeoHashGrid.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/GeoHashGrid.java index aa1d5504ad24f..9b6713ac033ae 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/GeoHashGrid.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/GeoHashGrid.java @@ -76,7 +76,7 @@ protected InternalGeoHashGridBucket createBucket(long hashAsLong, long docCount, } @Override - protected Reader getBucketReader() { + protected Reader getBucketReader() { return InternalGeoHashGridBucket::new; } diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/GeoTileGrid.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/GeoTileGrid.java index 91c523c80855e..bf45080759a07 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/GeoTileGrid.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/GeoTileGrid.java @@ -76,7 +76,7 @@ protected InternalGeoTileGridBucket createBucket(long hashAsLong, long docCount, } @Override - protected Reader getBucketReader() { + protected Reader getBucketReader() { return InternalGeoTileGridBucket::new; } diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGridBucket.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGridBucket.java index 493c84ed9dfd7..6e7ed8a679681 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGridBucket.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGridBucket.java @@ -33,7 +33,6 @@ import org.opensearch.common.geo.GeoPoint; import org.opensearch.common.io.stream.StreamInput; -import org.opensearch.core.common.io.stream.BaseStreamInput; import org.opensearch.geometry.utils.Geohash; import org.opensearch.search.aggregations.InternalAggregations; @@ -52,8 +51,8 @@ class InternalGeoHashGridBucket extends BaseGeoGridBucket C readOptionalNamedWriteable(Class category * @return the list of objects * @throws IOException if an I/O exception occurs reading the list */ - public List readList(final Writeable.Reader reader) throws IOException { + public List readList(final BaseWriteable.Reader reader) throws IOException { return readCollection(reader, ArrayList::new, Collections.emptyList()); } @@ -1207,8 +1207,11 @@ public Set readSet(Writeable.Reader reader) throws IOException { /** * Reads a collection of objects */ - private > C readCollection(Writeable.Reader reader, IntFunction constructor, C empty) - throws IOException { + private > C readCollection( + BaseWriteable.Reader reader, + IntFunction constructor, + C empty + ) throws IOException { int count = readArraySize(); if (count == 0) { return empty; diff --git a/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java b/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java index 3df6e9752c8b2..b0f4f6c8a6139 100644 --- a/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java +++ b/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java @@ -1176,7 +1176,7 @@ public void writeOptionalZoneId(@Nullable ZoneId timeZone) throws IOException { /** * Writes a collection to this stream. The corresponding collection can be read from a stream input using - * {@link StreamInput#readList(Writeable.Reader)}. + * {@link StreamInput#readList(BaseWriteable.Reader)}. * * @param collection the collection to write to this stream * @throws IOException if an I/O exception occurs writing the collection @@ -1207,7 +1207,7 @@ public void writeCollection(final Collection collection, final Writer /** * Writes a collection of a strings. The corresponding collection can be read from a stream input using - * {@link StreamInput#readList(Writeable.Reader)}. + * {@link StreamInput#readList(BaseWriteable.Reader)}. * * @param collection the collection of strings * @throws IOException if an I/O exception occurs writing the collection @@ -1218,7 +1218,7 @@ public void writeStringCollection(final Collection collection) throws IO /** * Writes an optional collection of a strings. The corresponding collection can be read from a stream input using - * {@link StreamInput#readList(Writeable.Reader)}. + * {@link StreamInput#readList(BaseWriteable.Reader)}. * * @param collection the collection of strings * @throws IOException if an I/O exception occurs writing the collection