ctubbsii commented on a change in pull request #2224: URL: https://github.com/apache/accumulo/pull/2224#discussion_r683781381
########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/EncodingOptions.java ########## @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.util.StringJoiner; + +/** + * Serialization metadata to allow for evolution of the encoding used for property storage. This + * info is expected to be stored first in the serialization and uncompressed so that the handling of + * subsequent fields and data can be processed correctly. + * <p> + * Instances of this class are immutable. + */ +public class EncodingOptions { + + public static final EncodingOptions COMPRESSED_V1 = + new EncodingOptions(EncodingOptions.EncodingVersion.V1_0, true); + + public static final EncodingOptions UNCOMPRESSED_V1 = + new EncodingOptions(EncodingOptions.EncodingVersion.V1_0, false); + + private final EncodingVersion encodingVersion; + private final boolean compress; + + public EncodingOptions(EncodingVersion encodingVersion, final boolean compress) { + this.encodingVersion = encodingVersion; + this.compress = compress; + } + + public EncodingOptions(final DataInputStream dis) throws IOException { + encodingVersion = EncodingVersion.byId(dis.readInt()); + compress = dis.readBoolean(); + } + + public void encode(final DataOutputStream dos) throws IOException { + dos.writeInt(encodingVersion.id); + dos.writeBoolean(compress); + } + + public EncodingVersion getEncodingVersion() { + return encodingVersion; + } + + public boolean isCompressed() { + return compress; + } + + @Override + public String toString() { + return new StringJoiner(", ", EncodingOptions.class.getSimpleName() + "[", "]") + .add("encodingVersion=" + encodingVersion).toString(); + } + + /** + * Provides a strong typing of the known encoding versions and allows the version id to be encoded + * as an integer. Adding an encoding type must be done as an addition and not change or delete + * previous versions or numbering to preserve compatibility. + */ + public enum EncodingVersion { Review comment: Not sure you need a whole class to represent a 1-up number. Can just use `int` instead of `EncodingVersion`. ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/GzipPropEncoding.java ########## @@ -0,0 +1,237 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +import static org.apache.accumulo.server.conf.codec.VersionedProperties.tsFormatter; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.time.Instant; +import java.util.AbstractMap; +import java.util.HashMap; +import java.util.Map; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +/** + * Initial property encoding that (optionally) uses gzip to compress the property map. The encoding + * version supported is EncodingVersion.V1_0. + */ +public class GzipPropEncoding implements PropSerdes { + + private final EncodingOptions encodingOpts; + + public GzipPropEncoding(final EncodingOptions encodingOpts) { + this.encodingOpts = encodingOpts; + } + + /** + * Serialize the versioned properties. The version information on the properties is updated if the + * data is successfully serialized. + * + * @param vProps + * the versioned properties. + * @return a byte array with the serialized properties. + */ + @Override + public byte[] toBytes(final VersionedProperties vProps) { + + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(); + DataOutputStream dos = new DataOutputStream(bos)) { + + // write header - version id, isCompressed + encodingOpts.encode(dos); + + // write updated property versioning info (data version, time stamp) + dos.writeInt(vProps.getNextVersion()); + dos.writeUTF(vProps.getTimestampISO()); Review comment: I'm unclear on the utility of storing a string timestamp. ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/PropSerdesEncoderFactory.java ########## @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +/** + * Create an serialization / deserialization encoder based on serialization version and additional + * information used in the serialization process. + */ +public class PropSerdesEncoderFactory { + + private final PropSerdes serdes; + + public PropSerdesEncoderFactory(final EncodingOptions encodingOptions) { + + // currently only one version supported - this would need to be extended to support others. Review comment: I'm not sure we need this factory at all. It's just extra layers of code to trace through and limits comprehension. We can just instantiate the GZip implementation directly. I don't anticipate making this pluggable or anything. ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/GzipPropEncoding.java ########## @@ -0,0 +1,237 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +import static org.apache.accumulo.server.conf.codec.VersionedProperties.tsFormatter; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.time.Instant; +import java.util.AbstractMap; +import java.util.HashMap; +import java.util.Map; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +/** + * Initial property encoding that (optionally) uses gzip to compress the property map. The encoding + * version supported is EncodingVersion.V1_0. + */ +public class GzipPropEncoding implements PropSerdes { + + private final EncodingOptions encodingOpts; + + public GzipPropEncoding(final EncodingOptions encodingOpts) { + this.encodingOpts = encodingOpts; + } + + /** + * Serialize the versioned properties. The version information on the properties is updated if the + * data is successfully serialized. + * + * @param vProps + * the versioned properties. + * @return a byte array with the serialized properties. + */ + @Override + public byte[] toBytes(final VersionedProperties vProps) { + + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(); + DataOutputStream dos = new DataOutputStream(bos)) { + + // write header - version id, isCompressed + encodingOpts.encode(dos); + + // write updated property versioning info (data version, time stamp) + dos.writeInt(vProps.getNextVersion()); Review comment: Is this version used to compare with the 1-up in ZK to ensure they are the same? If so, I'm not sure we need to store it inside the serialized blob, since it's only metadata about the blob. The VersionedProperty object should have it, certainly. I'm just not sure if it has any value stored inside the blob. ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/EncodingOptions.java ########## @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.util.StringJoiner; + +/** + * Serialization metadata to allow for evolution of the encoding used for property storage. This + * info is expected to be stored first in the serialization and uncompressed so that the handling of + * subsequent fields and data can be processed correctly. + * <p> + * Instances of this class are immutable. + */ +public class EncodingOptions { + + public static final EncodingOptions COMPRESSED_V1 = + new EncodingOptions(EncodingOptions.EncodingVersion.V1_0, true); + + public static final EncodingOptions UNCOMPRESSED_V1 = + new EncodingOptions(EncodingOptions.EncodingVersion.V1_0, false); + + private final EncodingVersion encodingVersion; + private final boolean compress; + + public EncodingOptions(EncodingVersion encodingVersion, final boolean compress) { + this.encodingVersion = encodingVersion; + this.compress = compress; + } + + public EncodingOptions(final DataInputStream dis) throws IOException { + encodingVersion = EncodingVersion.byId(dis.readInt()); + compress = dis.readBoolean(); + } + + public void encode(final DataOutputStream dos) throws IOException { + dos.writeInt(encodingVersion.id); + dos.writeBoolean(compress); + } + + public EncodingVersion getEncodingVersion() { + return encodingVersion; + } + + public boolean isCompressed() { + return compress; + } + + @Override + public String toString() { + return new StringJoiner(", ", EncodingOptions.class.getSimpleName() + "[", "]") + .add("encodingVersion=" + encodingVersion).toString(); + } Review comment: Not sure what this toString method is intended to be used for, but for completeness, could add compress field. ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/GzipPropEncoding.java ########## @@ -0,0 +1,237 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +import static org.apache.accumulo.server.conf.codec.VersionedProperties.tsFormatter; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.time.Instant; +import java.util.AbstractMap; +import java.util.HashMap; +import java.util.Map; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +/** + * Initial property encoding that (optionally) uses gzip to compress the property map. The encoding + * version supported is EncodingVersion.V1_0. + */ +public class GzipPropEncoding implements PropSerdes { + + private final EncodingOptions encodingOpts; + + public GzipPropEncoding(final EncodingOptions encodingOpts) { + this.encodingOpts = encodingOpts; + } + + /** + * Serialize the versioned properties. The version information on the properties is updated if the + * data is successfully serialized. + * + * @param vProps + * the versioned properties. + * @return a byte array with the serialized properties. + */ + @Override + public byte[] toBytes(final VersionedProperties vProps) { + + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(); + DataOutputStream dos = new DataOutputStream(bos)) { + + // write header - version id, isCompressed + encodingOpts.encode(dos); + + // write updated property versioning info (data version, time stamp) + dos.writeInt(vProps.getNextVersion()); + dos.writeUTF(vProps.getTimestampISO()); + + // write the property map keys, values. + if (encodingOpts.isCompressed()) { + writeMapCompressed(bos, vProps.getAllProperties()); + } else { + writeMap(dos, vProps.getAllProperties()); + } + + dos.flush(); + + return bos.toByteArray(); + + } catch (IOException ex) { + throw new IllegalStateException("Encountered error serializing properties", ex); + } + } + + @Override + public VersionedProperties fromBytes(final byte[] bytes) { + + try (ByteArrayInputStream bis = new ByteArrayInputStream(bytes); + DataInputStream dis = new DataInputStream(bis)) { + + // read header - version and isCompressed. + EncodingOptions opts = new EncodingOptions(dis); + + // read versioning information + int dataVersion = dis.readInt(); + Instant timestamp = tsFormatter.parse(dis.readUTF(), Instant::from); + + // read the property map keys, values + Map<String,String> aMap; + if (opts.isCompressed()) { + aMap = readCompressedMap(bis); + } else { + aMap = readMap(dis); + } + + return new VersionedPropertiesImpl(dataVersion, timestamp, aMap); + + } catch (IOException ex) { + throw new UncheckedIOException("Encountered error deserializing properties", ex); + } + } + + /** + * Read and uncompress an input stream compressed with GZip. The input stream is not closed by + * this method. + * + * @param is + * an input stream + * @return a map with the property k.v pairs + * @throws IOException + * if an exception occurs reading from the stream + */ + private Map<String,String> readCompressedMap(final InputStream is) throws IOException { + + try (GZIPInputStream gzipIn = new GZIPInputStream(is); + DataInputStream dis = new DataInputStream(gzipIn)) { + return readMap(dis); + } + } + + /** + * Read the property map from a data input stream + * + * @param dis + * a data input stream + * @return the property map + * @throws IOException + * if an exception occurs reading from the stream. + */ + private Map<String,String> readMap(DataInputStream dis) throws IOException { + + Map<String,String> aMap = new HashMap<>(); + int items = dis.readInt(); + + for (int i = 0; i < items; i++) { + Map.Entry<String,String> e = readKV(dis); + aMap.put(e.getKey(), e.getValue()); + } + return aMap; + } + + /** + * Write the property map to the output stream, compressing the output using GZip compression. The + * underlying stream will be closed when this method completes. + * + * @param os + * an output stream + * @param aMap + * the property map of k, v string pairs. + * @throws IOException + * if an exception occurs. + */ + private void writeMapCompressed(final OutputStream os, final Map<String,String> aMap) + throws IOException { + + try (GZIPOutputStream gzipOut = new GZIPOutputStream(os); + DataOutputStream dos = new DataOutputStream(gzipOut)) { + + writeMap(dos, aMap); + + gzipOut.flush(); + gzipOut.finish(); + + } catch (IOException ex) { + throw new IOException("Encountered error compressing properties", ex); + } + } + + /** + * Write the property map to the data output stream. The underlying stream is not closed by this + * method. + * + * @param dos + * a data output stream + * @param aMap + * the property map of k, v string pairs. + * @throws IOException + * if an exception occurs. + */ + private void writeMap(final DataOutputStream dos, final Map<String,String> aMap) + throws IOException { + + dos.writeInt(aMap.size()); + + aMap.forEach((k, v) -> writeKV(k, v, dos)); + + dos.flush(); Review comment: small nit. I'd clean up some of these blank lines. The more the code is spread out like that, the harder it is to fit on one screen, especially when using large fonts, making it harder to read (for me, at least) ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/PropSerdes.java ########## @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +/** + * Serialize / Deserialize versioned properties into a byte array. + */ +public interface PropSerdes { Review comment: Not sure I like the name of this interface, but the name isn't important. I think this is a good interface, and I like how the GZip implementation extends and implements it. ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/VersionedProperties.java ########## @@ -0,0 +1,137 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.util.Collection; +import java.util.Map; + +public interface VersionedProperties { Review comment: I don't think this needs to be an interface with a separate implementation. These two can be combined and simplified to just a simple POJO container of `{ version, propMap }`. I'm thinking something like: ```java public class VersionedProperties { private final long version; private final Map<String,String> properties; public VersionedProperties(long version, Map<String,String> properties) { this.version = version; this.properties = Map.copyOf(requireNonNull(properties)); } public long getVersion() { return version; } public Map<String,String> getProperties() { return properties; } } ``` ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/PropSerdesEncoderFactory.java ########## @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +/** + * Create an serialization / deserialization encoder based on serialization version and additional + * information used in the serialization process. + */ +public class PropSerdesEncoderFactory { + + private final PropSerdes serdes; + + public PropSerdesEncoderFactory(final EncodingOptions encodingOptions) { + + // currently only one version supported - this would need to be extended to support others. + if (encodingOptions.getEncodingVersion().equals(EncodingOptions.EncodingVersion.V1_0)) { Review comment: Should always compare enums with `==` and never `.equals()`. ```suggestion if (encodingOptions.getEncodingVersion() == EncodingOptions.EncodingVersion.V1_0) { ``` ########## File path: server/base/src/main/java/org/apache/accumulo/server/conf/codec/EncodingOptions.java ########## @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.conf.codec; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.util.StringJoiner; + +/** + * Serialization metadata to allow for evolution of the encoding used for property storage. This + * info is expected to be stored first in the serialization and uncompressed so that the handling of + * subsequent fields and data can be processed correctly. + * <p> + * Instances of this class are immutable. + */ +public class EncodingOptions { + + public static final EncodingOptions COMPRESSED_V1 = + new EncodingOptions(EncodingOptions.EncodingVersion.V1_0, true); + + public static final EncodingOptions UNCOMPRESSED_V1 = + new EncodingOptions(EncodingOptions.EncodingVersion.V1_0, false); + + private final EncodingVersion encodingVersion; + private final boolean compress; + + public EncodingOptions(EncodingVersion encodingVersion, final boolean compress) { + this.encodingVersion = encodingVersion; + this.compress = compress; + } + + public EncodingOptions(final DataInputStream dis) throws IOException { + encodingVersion = EncodingVersion.byId(dis.readInt()); + compress = dis.readBoolean(); + } + + public void encode(final DataOutputStream dos) throws IOException { + dos.writeInt(encodingVersion.id); + dos.writeBoolean(compress); + } + + public EncodingVersion getEncodingVersion() { + return encodingVersion; + } + + public boolean isCompressed() { + return compress; + } + + @Override + public String toString() { + return new StringJoiner(", ", EncodingOptions.class.getSimpleName() + "[", "]") + .add("encodingVersion=" + encodingVersion).toString(); + } + + /** + * Provides a strong typing of the known encoding versions and allows the version id to be encoded + * as an integer. Adding an encoding type must be done as an addition and not change or delete + * previous versions or numbering to preserve compatibility. + */ + public enum EncodingVersion { + + INVALID(-1), V1_0(1); + + // a unique id to identify the version + public final Integer id; Review comment: Just use primitive `int` here to avoid the unnecessary autoboxing/unboxing. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
