Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

JNI compression, options for compressors #269

Merged
merged 1 commit into from
May 16, 2024
Merged

JNI compression, options for compressors #269

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions managed-ledger/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,22 @@
<artifactId>protobuf-java</artifactId>
</dependency>

<dependency>
<groupId>org.lz4</groupId>
<artifactId>lz4-java</artifactId>
<version>1.5.0</version>
</dependency>

<dependency>
<groupId>com.github.luben</groupId>
<artifactId>zstd-jni</artifactId>
</dependency>

<dependency>
<groupId>org.xerial.snappy</groupId>
<artifactId>snappy-java</artifactId>
</dependency>

<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-common</artifactId>
Expand Down
6 changes: 3 additions & 3 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3302,7 +3302,7 @@ private byte[] compressDataIfNeeded(byte[] data, LedgerHandle lh) {
if (pulsarCursorInfoCompression != null) {
String pulsarCursorInfoCompressionString = new String(pulsarCursorInfoCompression);
CompressionCodec compressionCodec = CompressionCodecProvider.getCompressionCodec(
CompressionType.valueOf(pulsarCursorInfoCompressionString));
CompressionType.valueOf(pulsarCursorInfoCompressionString), data.length);
ByteBuf encode = compressionCodec.encode(Unpooled.wrappedBuffer(data));
try {
int uncompressedSize = data.length;
Expand Down Expand Up @@ -3331,13 +3331,13 @@ private static byte[] decompressDataIfNeeded(byte[] data, LedgerHandle lh) {
lh.getCustomMetadata().get(METADATA_PROPERTY_CURSOR_COMPRESSION_TYPE);
if (pulsarCursorInfoCompression != null) {
String pulsarCursorInfoCompressionString = new String(pulsarCursorInfoCompression);
CompressionCodec compressionCodec = CompressionCodecProvider.getCompressionCodec(
CompressionType.valueOf(pulsarCursorInfoCompressionString));
ByteArrayInputStream input = new ByteArrayInputStream(data);
DataInputStream dataInputStream = new DataInputStream(input);
try {
int uncompressedSize = dataInputStream.readInt();
byte[] compressedData = dataInputStream.readNBytes(uncompressedSize);
CompressionCodec compressionCodec = CompressionCodecProvider.getCompressionCodec(
CompressionType.valueOf(pulsarCursorInfoCompressionString), uncompressedSize);
ByteBuf decode = compressionCodec.decode(Unpooled.wrappedBuffer(compressedData), uncompressedSize);
try {
return ByteBufUtil.getBytes(decode);
Expand Down
21 changes: 11 additions & 10 deletions pulsar-common/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,30 +201,31 @@
<artifactId>protobuf-java</artifactId>
</dependency>

<!-- test -->
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bc-fips</artifactId>
<version>${bouncycastle.bc-fips.version}</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>org.lz4</groupId>
<artifactId>lz4-java</artifactId>
<version>1.5.0</version>
<scope>test</scope>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>com.github.luben</groupId>
<artifactId>zstd-jni</artifactId>
<scope>test</scope>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>org.xerial.snappy</groupId>
<artifactId>snappy-java</artifactId>
<scope>provided</scope>
</dependency>

<!-- test -->

<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bc-fips</artifactId>
<version>${bouncycastle.bc-fips.version}</version>
<scope>test</scope>
</dependency>

Expand Down
10 changes: 9 additions & 1 deletion ...n/compression/CompressionCodecLZ4JNI.java → ...n/compression/CompressionCodecLZ4JNI.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,16 @@ public class CompressionCodecLZ4JNI implements CompressionCodec {
}
}

private static final boolean useLz4HighCompression = Boolean.parseBoolean(System.
getProperty("pulsar.compression.lz4JniUseHighCompression", "false"));
// should be in the range of [1, 17]
private static final int lz4CompressionLevel = Integer.parseInt(System.
getProperty("pulsar.compression.lz4JniHighCompressionLevel", "9"));

private static final LZ4Factory lz4Factory = LZ4Factory.fastestInstance();
private static final LZ4Compressor compressor = lz4Factory.fastCompressor();
private static final LZ4Compressor compressor = useLz4HighCompression
? lz4Factory.highCompressor(lz4CompressionLevel)
: lz4Factory.fastCompressor();
private static final LZ4FastDecompressor decompressor = lz4Factory.fastDecompressor();

@Override
Expand Down
43 changes: 42 additions & 1 deletion ...r-common/src/main/java/org/apache/pulsar/common/compression/CompressionCodecProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import java.util.EnumMap;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.CompressionType;

/**
Expand All @@ -32,8 +33,12 @@
* @see CompressionCodecSnappy
*/
@UtilityClass
@Slf4j
public class CompressionCodecProvider {
private static final EnumMap<org.apache.pulsar.common.api.proto.CompressionType, CompressionCodec> codecs;
private static final EnumMap<org.apache.pulsar.common.api.proto.CompressionType, CompressionCodec> jniCodecs;

private static final int jniRecommendedSize = jniRecommendedMinSize();

static {
codecs = new EnumMap<>(org.apache.pulsar.common.api.proto.CompressionType.class);
Expand All @@ -42,16 +47,52 @@ public class CompressionCodecProvider {
codecs.put(org.apache.pulsar.common.api.proto.CompressionType.ZLIB, new CompressionCodecZLib());
codecs.put(org.apache.pulsar.common.api.proto.CompressionType.ZSTD, new CompressionCodecZstd());
codecs.put(org.apache.pulsar.common.api.proto.CompressionType.SNAPPY, new CompressionCodecSnappy());

jniCodecs = new EnumMap<>(org.apache.pulsar.common.api.proto.CompressionType.class);
if (allowJni()) {
try {
// avoid exceptions from lz4 being swallowed until its use
net.jpountz.util.Native.load();

jniCodecs.put(org.apache.pulsar.common.api.proto.CompressionType.LZ4, new CompressionCodecLZ4JNI());
jniCodecs.put(org.apache.pulsar.common.api.proto.CompressionType.ZSTD, new CompressionCodecZstdJNI());
jniCodecs.put(org.apache.pulsar.common.api.proto.CompressionType.SNAPPY,
new CompressionCodecSnappyJNI());
} catch (Throwable t) {
log.error("Failed to load JNI compression codecs", t);
}
}
}

public static boolean allowJni() {
return System.getProperty("pulsar.compression.allowJni", "false").equals("true");
}

public static int jniRecommendedMinSize() {
return Integer.parseInt(System.getProperty("pulsar.compression.jniMinSize", "10240"));
}

public static CompressionCodec getCompressionCodec(org.apache.pulsar.common.api.proto.CompressionType type) {
return codecs.get(type);
}

public static CompressionCodec getCompressionCodec(CompressionType type, int size) {
return getCompressionCodec(convertToWireProtocol(type), size);
}

public static CompressionCodec getCompressionCodec(org.apache.pulsar.common.api.proto.CompressionType type,
int size) {
if (size > jniRecommendedSize && jniCodecs.containsKey(type)) {
return jniCodecs.get(type);
}
return codecs.get(type);
}

public static CompressionCodec getCompressionCodec(CompressionType type) {
return codecs.get(convertToWireProtocol(type));
return getCompressionCodec(convertToWireProtocol(type));
}


public static org.apache.pulsar.common.api.proto.CompressionType convertToWireProtocol(
CompressionType compressionType) {
switch (compressionType) {
Expand Down
0 ...ompression/CompressionCodecSnappyJNI.java → ...ompression/CompressionCodecSnappyJNI.java
View file
Open in desktop
File renamed without changes.
3 changes: 2 additions & 1 deletion pulsar-common/src/main/java/org/apache/pulsar/common/compression/CompressionCodecZstd.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,8 @@
*/
public class CompressionCodecZstd implements CompressionCodec {

private static final int ZSTD_COMPRESSION_LEVEL = 3;
private static final int ZSTD_COMPRESSION_LEVEL = Integer.parseInt(System.
getProperty("pulsar.compression.zstdCompressionLevel", "3"));

private static final ZstdCompressor ZSTD_COMPRESSOR = new ZstdCompressor();

Expand Down
6 changes: 2 additions & 4 deletions .../compression/CompressionCodecZstdJNI.java → .../compression/CompressionCodecZstdJNI.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,20 +19,18 @@
package org.apache.pulsar.common.compression;

import com.github.luben.zstd.Zstd;

import io.netty.buffer.ByteBuf;

import java.io.IOException;
import java.nio.ByteBuffer;

import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;

/**
* Zstandard Compression.
*/
public class CompressionCodecZstdJNI implements CompressionCodec {

private static final int ZSTD_COMPRESSION_LEVEL = 3;
private static final int ZSTD_COMPRESSION_LEVEL = Integer.parseInt(System.
getProperty("pulsar.compression.zstdCompressionLevel", "3"));

@Override
public ByteBuf encode(ByteBuf source) {
Expand Down
Loading