[pulsar-common] Support Snappy compression for Java (apache#4259)

* Support Snappy compression for java.

* Some minor fix to pass unit tests

* Format the cpp code

* Added support for c++ client

* Format the cpp code

build/docker/Dockerfile

@@ -28,7 +28,7 @@ RUN apt-get update && \
                 liblog4cxx-dev libprotobuf-dev libboost-all-dev google-mock libgtest-dev \
                 libjsoncpp-dev libxml2-utils protobuf-compiler wget \
                 curl doxygen openjdk-8-jdk-headless clang-format-5.0 \
-                gnupg2 golang-1.10-go zip unzip libzstd-dev
+                gnupg2 golang-1.10-go zip unzip libzstd-dev libsnappy-dev
 
 # Compile and install gtest
 RUN cd /usr/src/gtest && cmake . && make && cp libgtest.a /usr/lib

pom.xml

@@ -194,6 +194,7 @@ flexible messaging model and an intuitive client API.</description>
     <jsonwebtoken.version>0.10.5</jsonwebtoken.version>
     <opencensus.version>0.18.0</opencensus.version>
     <zstd.version>1.3.7-3</zstd.version>
+    <snappy.version>1.1.1.3</snappy.version>
     <hbase.version>1.4.9</hbase.version>
 
     <!-- test dependencies -->
@@ -487,6 +488,12 @@ flexible messaging model and an intuitive client API.</description>
         <version>${zstd.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>org.xerial.snappy</groupId>
+        <artifactId>snappy-java</artifactId>
+        <version>${snappy.version}</version>
+      </dependency>
+
       <dependency>
         <groupId>org.slf4j</groupId>
         <artifactId>slf4j-api</artifactId>

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/CompressionType.java

@@ -33,4 +33,7 @@ public enum CompressionType {
 
     /** Compress with Zstandard codec */
     ZSTD,
+
+    /** Compress with Snappy codec */
+    SNAPPY
 }

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java

@@ -260,6 +260,8 @@ public interface ProducerBuilder<T> extends Cloneable {
      * <li>{@link CompressionType#ZLIB}: Standard ZLib compression</li>
      * <li>{@link CompressionType#ZSTD} Compress with Zstandard codec. Since Pulsar 2.3. Zstd cannot be used if consumer
      * applications are not in version >= 2.3 as well</li>
+     * <li>{@link CompressionType#SNAPPY} Compress with Snappy codec. Since Pulsar 2.4. Snappy cannot be used if consumer
+     * applications are not in version >= 2.4 as well</li>
      * </ul>
      *
      * @param compressionType

pulsar-client-cpp/CMakeLists.txt

@@ -120,6 +120,8 @@ endif()
 
 find_library(LIB_ZSTD zstd libzstd)
 
+find_library(LIB_SNAPPY snappy libsnappy)
+
 if (BUILD_PYTHON_WRAPPER)
     find_package(PythonLibs REQUIRED)
     MESSAGE(STATUS "PYTHON: " ${PYTHONLIBS_VERSION_STRING})
@@ -183,6 +185,12 @@ else ()
 endif ()
 MESSAGE(STATUS "HAS_ZSTD: ${HAS_ZSTD}")
 
+if (LIB_SNAPPY)
+    set(HAS_SNAPPY 1)
+else ()
+    set(HAS_SNAPPY 0)
+endif ()
+MESSAGE(STATUS "HAS_SNAPPY: ${HAS_SNAPPY}")
 
 set(ADDITIONAL_LIBRARIES $ENV{PULSAR_ADDITIONAL_LIBRARIES})
 link_directories( $ENV{PULSAR_ADDITIONAL_LIBRARY_PATH} )
@@ -241,6 +249,12 @@ endif ()
 
 add_definitions(-DHAS_ZSTD=${HAS_ZSTD})
 
+if (HAS_SNAPPY)
+    set(COMMON_LIBS ${COMMON_LIBS} ${LIB_SNAPPY} )
+endif ()
+
+add_definitions(-DHAS_SNAPPY=${HAS_SNAPPY})
+
 if(NOT APPLE AND NOT MSVC)
     set(COMMON_LIBS ${COMMON_LIBS} rt)
 endif ()

pulsar-client-cpp/docker/Dockerfile

@@ -149,6 +149,14 @@ RUN curl -O -L https://github.com/facebook/zstd/releases/download/v1.3.7/zstd-1.
     make install && \
     rm -rf /zstd-1.3.7 /zstd-1.3.7.tar.gz
 
+# Snappy
+RUN curl -O -L https://github.com/google/snappy/releases/download/1.1.3/snappy-1.1.3.tar.gz && \
+    tar xvfz snappy-1.1.3.tar.gz && \
+    cd snappy-1.1.3 && \
+    CFLAGS="-fPIC -O3" ./configure && \
+    make && make install && \
+    rm -rf /snappy-1.1.3 /snappy-1.1.3.tar.gz
+
 RUN pip install twine
 RUN pip install fastavro
 RUN pip install six

pulsar-client-cpp/include/pulsar/CompressionType.h

@@ -26,6 +26,7 @@ enum CompressionType
     CompressionLZ4 = 1,
     CompressionZLib = 2,
     CompressionZSTD = 3,
+    CompressionSNAPPY = 4
 };
 }
 

pulsar-client-cpp/include/pulsar/ProducerConfiguration.h

@@ -100,6 +100,8 @@ class PULSAR_PUBLIC ProducerConfiguration {
      * <li>{@link CompressionZLib}: ZLib Compression http://zlib.net/</li>
      * <li>{@link CompressionZSTD}: Zstandard Compression  https://facebook.github.io/zstd/ (Since Pulsar 2.3.
      * Zstd cannot be used if consumer applications are not in version >= 2.3 as well)</li>
+     * <li>{@link CompressionSNAPPY}: Snappy Compression  https://google.github.io/snappy/ (Since Pulsar 2.4.
+     * Snappy cannot be used if consumer applications are not in version >= 2.4 as well)</li>
      * </ul>
      */
     ProducerConfiguration& setCompressionType(CompressionType compressionType);

pulsar-client-cpp/lib/CompressionCodec.cc

@@ -20,6 +20,7 @@
 #include "CompressionCodecLZ4.h"
 #include "CompressionCodecZLib.h"
 #include "CompressionCodecZstd.h"
+#include "CompressionCodecSnappy.h"
 
 #include <cassert>
 
@@ -30,6 +31,7 @@ CompressionCodecNone CompressionCodecProvider::compressionCodecNone_;
 CompressionCodecLZ4 CompressionCodecProvider::compressionCodecLZ4_;
 CompressionCodecZLib CompressionCodecProvider::compressionCodecZLib_;
 CompressionCodecZstd CompressionCodecProvider::compressionCodecZstd_;
+CompressionCodecSnappy CompressionCodecProvider::compressionCodecSnappy_;
 
 CompressionCodec& CompressionCodecProvider::getCodec(CompressionType compressionType) {
     switch (compressionType) {
@@ -39,6 +41,8 @@ CompressionCodec& CompressionCodecProvider::getCodec(CompressionType compression
             return compressionCodecZLib_;
         case CompressionZSTD:
             return compressionCodecZstd_;
+        case CompressionSNAPPY:
+            return compressionCodecSnappy_;
         default:
             return compressionCodecNone_;
     }
@@ -54,6 +58,8 @@ CompressionType CompressionCodecProvider::convertType(proto::CompressionType typ
             return CompressionZLib;
         case proto::ZSTD:
             return CompressionZSTD;
+        case proto::SNAPPY:
+            return CompressionSNAPPY;
     }
 }
 
@@ -67,6 +73,8 @@ proto::CompressionType CompressionCodecProvider::convertType(CompressionType typ
             return proto::ZLIB;
         case CompressionZSTD:
             return proto::ZSTD;
+        case CompressionSNAPPY:
+            return proto::SNAPPY;
     }
 }
 

pulsar-client-cpp/lib/CompressionCodec.h

@@ -35,6 +35,7 @@ class CompressionCodecNone;
 class CompressionCodecLZ4;
 class CompressionCodecZLib;
 class CompressionCodecZstd;
+class CompressionCodecSnappy;
 
 class PULSAR_PUBLIC CompressionCodecProvider {
    public:
@@ -48,6 +49,7 @@ class PULSAR_PUBLIC CompressionCodecProvider {
     static CompressionCodecLZ4 compressionCodecLZ4_;
     static CompressionCodecZLib compressionCodecZLib_;
     static CompressionCodecZstd compressionCodecZstd_;
+    static CompressionCodecSnappy compressionCodecSnappy_;
 };
 
 class PULSAR_PUBLIC CompressionCodec {

pulsar-client-cpp/lib/CompressionCodecSnappy.cc

@@ -0,0 +1,79 @@
+/**
+ * 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.
+ */
+#include "CompressionCodecSnappy.h"
+
+#if HAS_SNAPPY
+#include <snappy.h>
+#include "snappy-c.h"
+
+namespace pulsar {
+
+SharedBuffer CompressionCodecSnappy::encode(const SharedBuffer& raw) {
+    // Get the max size of the compressed data and allocate a buffer to hold it
+    int maxCompressedSize = snappy_max_compressed_length(raw.readableBytes());
+    SharedBuffer compressed = SharedBuffer::allocate(maxCompressedSize);
+
+    unsigned long bytesWritten = maxCompressedSize;
+
+    snappy_status status =
+        snappy_compress(raw.data(), raw.readableBytes(), compressed.mutableData(), &bytesWritten);
+
+    if (status != SNAPPY_OK) {
+        LOG_ERROR("Failed to compress to snappy. res=" << res);
+        abort();
+    }
+
+    compressed.bytesWritten(bytesWritten);
+
+    return compressed;
+}
+
+bool CompressionCodecSnappy::decode(const SharedBuffer& encoded, uint32_t uncompressedSize,
+                                    SharedBuffer& decoded) {
+    SharedBuffer decompressed = SharedBuffer::allocate(uncompressedSize);
+
+    snappy_status status = snappy_uncompress(encoded.data(), encoded.readableBytes(),
+                                             decompressed.mutableData(), uncompressedSize);
+
+    if (status == SNAPPY_OK) {
+        decoded = decompressed;
+        decoded.setWriterIndex(uncompressedSize);
+        return true;
+    } else {
+        // Decompression failed
+        return false;
+    }
+}
+}  // namespace pulsar
+
+#else  // No SNAPPY
+
+namespace pulsar {
+
+SharedBuffer CompressionCodecSnappy::encode(const SharedBuffer& raw) {
+    throw "Snappy compression not supported";
+}
+
+bool CompressionCodecSnappy::decode(const SharedBuffer& encoded, uint32_t uncompressedSize,
+                                    SharedBuffer& decoded) {
+    throw "Snappy compression not supported";
+}
+}  // namespace pulsar
+
+#endif  // HAS_SNAPPY

pulsar-client-cpp/lib/CompressionCodecSnappy.h

@@ -0,0 +1,31 @@
+/**
+ * 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.
+ */
+#pragma once
+
+#include "CompressionCodec.h"
+
+namespace pulsar {
+
+class CompressionCodecSnappy : public CompressionCodec {
+   public:
+    SharedBuffer encode(const SharedBuffer& raw);
+
+    bool decode(const SharedBuffer& encoded, uint32_t uncompressedSize, SharedBuffer& decoded);
+};
+}  // namespace pulsar

pulsar-common/pom.xml

@@ -83,6 +83,11 @@
       <groupId>com.github.luben</groupId>
       <artifactId>zstd-jni</artifactId>
     </dependency>
+
+    <dependency>
+      <groupId>org.xerial.snappy</groupId>
+      <artifactId>snappy-java</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>org.bouncycastle</groupId>

pulsar-common/src/main/java/org/apache/pulsar/common/compression/CompressionCodecProvider.java

@@ -35,6 +35,7 @@ public class CompressionCodecProvider {
         codecs.put(PulsarApi.CompressionType.LZ4, new CompressionCodecLZ4());
         codecs.put(PulsarApi.CompressionType.ZLIB, new CompressionCodecZLib());
         codecs.put(PulsarApi.CompressionType.ZSTD, new CompressionCodecZstd());
+        codecs.put(PulsarApi.CompressionType.SNAPPY, new CompressionCodecSnappy());
     }
 
     public static CompressionCodec getCompressionCodec(PulsarApi.CompressionType type) {
@@ -55,6 +56,8 @@ public static PulsarApi.CompressionType convertToWireProtocol(CompressionType co
             return PulsarApi.CompressionType.ZLIB;
         case ZSTD:
             return PulsarApi.CompressionType.ZSTD;
+        case SNAPPY:
+            return PulsarApi.CompressionType.SNAPPY;
 
         default:
             throw new RuntimeException("Invalid compression type");
@@ -71,6 +74,8 @@ public static CompressionType convertFromWireProtocol(PulsarApi.CompressionType
             return CompressionType.ZLIB;
         case ZSTD:
             return CompressionType.ZSTD;
+        case SNAPPY:
+            return CompressionType.SNAPPY;
 
         default:
             throw new RuntimeException("Invalid compression type");