adoroszlai commented on code in PR #4089:
URL: https://github.com/apache/ozone/pull/4089#discussion_r1049664898
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/OnDemandContainerReplicationSource.java:
##########
@@ -37,11 +43,16 @@
private final ContainerController controller;
- private final TarContainerPacker packer = new TarContainerPacker();
+ private Map<String, TarContainerPacker> packer = new HashMap<>();
public OnDemandContainerReplicationSource(
ContainerController controller) {
this.controller = controller;
+ this.packer.put("NO_COMPRESSION", new
TarContainerPacker("no_compression"));
+ this.packer.put("GZIP", new TarContainerPacker(GZIP));
+ this.packer.put("LZ4", new TarContainerPacker(LZ4_FRAMED));
+ this.packer.put("SNAPPY", new TarContainerPacker(SNAPPY_FRAMED));
+ this.packer.put("ZSTD", new TarContainerPacker(ZSTANDARD));
Review Comment:
I think we should define a public Ozone compression name -> commons-compress
name mapping. It can be used to create `TarContainerPacker` on-demand, instead
of creating one for each compression type.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java:
##########
@@ -49,11 +49,13 @@ public GrpcReplicationService(ContainerReplicationSource
source) {
public void download(CopyContainerRequestProto request,
StreamObserver<CopyContainerResponseProto> responseObserver) {
long containerID = request.getContainerID();
- LOG.info("Streaming container data ({}) to other datanode", containerID);
+ String compression = request.getCompression().toString();
Review Comment:
`compression` may be absent, so we need to check:
```suggestion
String compression = request.hasCompression()
? request.getCompression().toString()
: "GZIP";
```
Also, please handle unknown compression by falling back to `NO_COMPRESSION`.
##########
hadoop-ozone/dist/src/main/license/jar-report.txt:
##########
@@ -265,3 +265,4 @@ share/ozone/lib/token-provider.jar
share/ozone/lib/txw2.jar
share/ozone/lib/weld-servlet.Final.jar
share/ozone/lib/woodstox-core.jar
+share/ozone/lib/zstd-jni.jar
Review Comment:
Please update `hadoop-ozone/dist/src/main/license/bin/LICENSE.txt` as well
with the new dependency.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/TarContainerPacker.java:
##########
@@ -60,6 +60,42 @@
private static final String CONTAINER_FILE_NAME = "container.yaml";
+ private final String compression;
+
+ private static final String NO_COMPRESSION = "no_compression";
+
+ public TarContainerPacker() {
+ this.compression = NO_COMPRESSION;
+ }
+
+ public TarContainerPacker(String compression) {
+ this.compression = compression;
+ }
+
+ private ArchiveInputStream getArchiveInputStream(InputStream input)
+ throws CompressorException {
+ ArchiveInputStream archiveInput;
+ if (compression == NO_COMPRESSION) {
+ archiveInput = untar(input);
+ } else {
+ InputStream decompressed = decompress(input);
+ archiveInput = untar(decompressed);
+ }
+ return archiveInput;
+ }
+
+ private ArchiveOutputStream getArchiveOutputStream(OutputStream output)
+ throws CompressorException {
+ ArchiveOutputStream archiveOutput;
+ if (compression == NO_COMPRESSION) {
+ archiveOutput = tar(output);
+ } else {
+ OutputStream compressed = compress(output);
+ archiveOutput = tar(compressed);
+ }
+ return archiveOutput;
+ }
Review Comment:
We can avoid some duplication here by making `compress/decompress` return
the argument for `NO_COMPRESSION` case. Also, please use `Objects.equals()`
instead of `==`.
```java
private InputStream decompress(InputStream input)
throws CompressorException {
return Objects.equals(compression, NO_COMPRESSION)
? input
: new CompressorStreamFactory()
.createCompressorInputStream(compression, input);
}
private OutputStream compress(OutputStream output)
throws CompressorException {
return Objects.equals(compression, NO_COMPRESSION)
? output
: new CompressorStreamFactory()
.createCompressorOutputStream(compression, output);
}
```
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java:
##########
@@ -67,6 +71,8 @@ public SimpleContainerDownloader(ConfigurationSource conf,
}
securityConfig = new SecurityConfig(conf);
this.certClient = certClient;
+ this.compression = conf.get(HDDS_CONTAINER_REPLICATION_COMPRESSION,
+ HDDS_CONTAINER_REPLICATION_COMPRESSION_DEFAULT);
Review Comment:
Please handle unknown compression by falling back to `NO_COMPRESSION`.
##########
hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java:
##########
@@ -211,7 +211,31 @@ public void testEmptyContainerImportExport() throws
Exception {
}
@Test
- public void testContainerImportExport() throws Exception {
+ public void testContainerImportExportUncompressed() throws Exception {
+ testContainerImportExport("no_compression");
+ }
+
+ @Test
+ public void testContainerImportExportGzip() throws Exception {
+ testContainerImportExport("gz");
+ }
+
+ @Test
+ public void testContainerImportExportZstd() throws Exception {
+ testContainerImportExport("zstd");
+ }
+
+ @Test
+ public void testContainerImportExportSnappy() throws Exception {
+ testContainerImportExport("snappy-framed");
+ }
+
+ @Test
+ public void testContainerImportExportLz4() throws Exception {
+ testContainerImportExport("lz4-framed");
+ }
Review Comment:
Use the Ozone compression name -> commons-compress name mapping to define
test cases dynamically.
##########
hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java:
##########
@@ -90,17 +93,32 @@ public class TestTarContainerPacker {
private final ContainerLayoutVersion layout;
private final String schemaVersion;
private OzoneConfiguration conf;
+ private String compression;
- public TestTarContainerPacker(ContainerTestVersionInfo versionInfo) {
+
+ public TestTarContainerPacker(
+ ContainerTestVersionInfo versionInfo, String compression) {
this.layout = versionInfo.getLayout();
this.schemaVersion = versionInfo.getSchemaVersion();
this.conf = new OzoneConfiguration();
ContainerTestVersionInfo.setTestSchemaVersion(schemaVersion, conf);
+ this.compression = compression;
+ packer = new TarContainerPacker(compression);
+
}
@Parameterized.Parameters
public static Iterable<Object[]> parameters() {
- return ContainerTestVersionInfo.versionParameters();
+ List<ContainerTestVersionInfo> layoutList =
+ ContainerTestVersionInfo.getLayoutList();
+ List<Object[]> parameterList = new ArrayList<>();
+ for (ContainerTestVersionInfo containerTestVersionInfo : layoutList) {
+ parameterList.add(new Object[]{containerTestVersionInfo, GZIP});
+ parameterList.add(new Object[]{containerTestVersionInfo, LZ4_FRAMED});
+ parameterList.add(new Object[]{containerTestVersionInfo, SNAPPY_FRAMED});
+ parameterList.add(new Object[]{containerTestVersionInfo, ZSTANDARD});
Review Comment:
Iterate the values of Ozone compression name -> commons-compress name
mapping. Make sure to include `no_compression`. Reuse
`TarContainerPacker.compress/decompress` for handling that case.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
