You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/01/19 01:16:16 UTC
[GitHub] merlimat closed pull request #1078: Re-implement DoubleByteBuf as a simple holder of a pair of buffers
merlimat closed pull request #1078: Re-implement DoubleByteBuf as a simple holder of a pair of buffers
URL: https://github.com/apache/incubator-pulsar/pull/1078
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index 20e6efb44..821061afe 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -66,7 +66,7 @@
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
import org.apache.bookkeeper.mledger.util.Pair;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
-import org.apache.pulsar.common.api.DoubleByteBuf;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream;
import org.apache.zookeeper.CreateMode;
@@ -2133,7 +2133,7 @@ public ByteBuf getMessageWithMetadata(byte[] data) throws IOException {
headers.writeInt(msgMetadataSize);
messageData.writeTo(outStream);
outStream.recycle();
- return DoubleByteBuf.get(headers, payload);
+ return ByteBufPair.coalesce(ByteBufPair.get(headers, payload));
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
index 3dee8fc04..cd0415aab 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
@@ -21,6 +21,7 @@
import java.io.File;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.PulsarDecoder;
import io.netty.channel.ChannelInitializer;
@@ -70,6 +71,8 @@ protected void initChannel(SocketChannel ch) throws Exception {
SslContext sslCtx = builder.clientAuth(ClientAuth.OPTIONAL).build();
ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
}
+
+ ch.pipeline().addLast("ByteBufPairEncoder", ByteBufPair.ENCODER);
ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4));
ch.pipeline().addLast("handler", new ServerCnx(brokerService));
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawMessage.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawMessage.java
index a68204c34..cc983e823 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawMessage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawMessage.java
@@ -18,10 +18,11 @@
*/
package org.apache.pulsar.client.api;
-import io.netty.buffer.ByteBuf;
-import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
+import io.netty.buffer.ByteBuf;
+
/**
* A representation of a message in a topic in its raw form (i.e. as it is stored in a managed ledger).
* RawMessages hold a refcount to the contains ByteBuf, so they must be closed for the ByteBuf to be freed.
@@ -45,10 +46,10 @@
ByteBuf getHeadersAndPayload();
/**
- * Serialize a raw message to a ByteBuf. The caller is responsible for releasing
- * the returned ByteBuf.
+ * Serialize a raw message to a ByteBufPair. The caller is responsible for releasing
+ * the returned ByteBufPair.
*/
- ByteBuf serialize();
+ ByteBufPair serialize();
@Override
void close();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java
index 2134e96bc..7160cc3be 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java
@@ -20,20 +20,18 @@
import java.io.IOException;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
-
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.RawMessage;
-import org.apache.pulsar.common.api.DoubleByteBuf;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
-import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream;
import org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+
public class RawMessageImpl implements RawMessage {
private static final Logger log = LoggerFactory.getLogger(RawMessageImpl.class);
@@ -67,7 +65,7 @@ public void close() {
}
@Override
- public ByteBuf serialize() {
+ public ByteBufPair serialize() {
// Format: [IdSize][Id][PayloadAndMetadataSize][PayloadAndMetadata]
int idSize = id.getSerializedSize();
int headerSize = 4 /* IdSize */ + idSize + 4 /* PayloadAndMetadataSize */;
@@ -85,7 +83,7 @@ public ByteBuf serialize() {
}
headers.writeInt(headersAndPayload.readableBytes());
- return DoubleByteBuf.get(headers, headersAndPayload);
+ return ByteBufPair.get(headers, headersAndPayload);
}
static public RawMessage deserializeFrom(ByteBuf buffer) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
index 62a6d690c..a5a561ddd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
@@ -39,7 +39,7 @@
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.broker.service.persistent.PersistentMessageExpiryMonitor;
import org.apache.pulsar.broker.service.persistent.PersistentMessageFinder;
-import org.apache.pulsar.common.api.DoubleByteBuf;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream;
import org.testng.annotations.Test;
@@ -68,7 +68,7 @@
ByteBufCodedOutputStream outStream = ByteBufCodedOutputStream.get(headers);
headers.writeInt(msgMetadataSize);
messageMetadata.writeTo(outStream);
- ByteBuf headersAndPayload = DoubleByteBuf.get(headers, data);
+ ByteBuf headersAndPayload = ByteBufPair.coalesce(ByteBufPair.get(headers, data));
byte[] byteMessage = headersAndPayload.nioBuffer().array();
headersAndPayload.release();
return byteMessage;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 7e7e9fc21..058a4dae5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -73,6 +73,7 @@
import org.apache.pulsar.broker.service.ServerCnx.State;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.utils.ClientChannelHelper;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.PulsarHandler;
import org.apache.pulsar.common.api.Commands.ChecksumType;
@@ -158,7 +159,7 @@ public void setup() throws Exception {
doReturn(Optional.empty()).when(zkDataCache).get(anyObject());
doReturn(zkDataCache).when(configCacheService).policiesCache();
doReturn(configCacheService).when(pulsar).getConfigurationCache();
-
+
LocalZooKeeperCacheService zkCache = mock(LocalZooKeeperCacheService.class);
doReturn(CompletableFuture.completedFuture(Optional.empty())).when(zkDataCache).getAsync(any());
doReturn(zkDataCache).when(zkCache).policiesCache();
@@ -571,7 +572,7 @@ public void testSendCommand() throws Exception {
.setProducerName("prod-name").setSequenceId(0).build();
ByteBuf data = Unpooled.buffer(1024);
- clientCommand = Commands.newSend(1, 0, 1, ChecksumType.None, messageMetadata, data);
+ clientCommand = ByteBufPair.coalesce(Commands.newSend(1, 0, 1, ChecksumType.None, messageMetadata, data));
channel.writeInbound(Unpooled.copiedBuffer(clientCommand));
clientCommand.release();
@@ -1292,7 +1293,7 @@ public void testSendSuccessOnEncryptionRequiredTopic() throws Exception {
.build();
ByteBuf data = Unpooled.buffer(1024);
- clientCommand = Commands.newSend(1, 0, 1, ChecksumType.None, messageMetadata, data);
+ clientCommand = ByteBufPair.coalesce(Commands.newSend(1, 0, 1, ChecksumType.None, messageMetadata, data));
channel.writeInbound(Unpooled.copiedBuffer(clientCommand));
clientCommand.release();
assertTrue(getResponse() instanceof CommandSendReceipt);
@@ -1326,7 +1327,7 @@ public void testSendFailureOnEncryptionRequiredTopic() throws Exception {
.build();
ByteBuf data = Unpooled.buffer(1024);
- clientCommand = Commands.newSend(1, 0, 1, ChecksumType.None, messageMetadata, data);
+ clientCommand = ByteBufPair.coalesce(Commands.newSend(1, 0, 1, ChecksumType.None, messageMetadata, data));
channel.writeInbound(Unpooled.copiedBuffer(clientCommand));
clientCommand.release();
assertTrue(getResponse() instanceof CommandSendError);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java
index 8f9801efb..900fadef6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java
@@ -53,6 +53,7 @@
import org.apache.pulsar.client.impl.ProducerImpl.OpSendMsg;
import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.Commands.ChecksumType;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata.Builder;
import org.slf4j.Logger;
@@ -471,7 +472,7 @@ public void testCorruptMessageRemove() throws Exception {
Builder metadataBuilder = ((MessageImpl) msg).getMessageBuilder();
MessageMetadata msgMetadata = metadataBuilder.setProducerName("test").setSequenceId(1).setPublishTime(10L)
.build();
- ByteBuf cmd = Commands.newSend(producerId, 1, 1, ChecksumType.Crc32c, msgMetadata, payload);
+ ByteBufPair cmd = Commands.newSend(producerId, 1, 1, ChecksumType.Crc32c, msgMetadata, payload);
// (a) create OpSendMsg with message-data : "message-1"
OpSendMsg op = OpSendMsg.create(((MessageImpl) msg), cmd, 1, null);
// a.verify: as message is not corrupt: no need to update checksum
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
index a80f25c3a..035bbf973 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
@@ -19,18 +19,17 @@
package org.apache.pulsar.client.impl;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
-
import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import org.testng.Assert;
import org.testng.annotations.Test;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+
public class RawMessageSerDeserTest {
static final Logger log = LoggerFactory.getLogger(RawMessageSerDeserTest.class);
@@ -45,7 +44,7 @@ public void testSerializationAndDeserialization() throws Exception {
.setPartition(10).setBatchIndex(20).build();
RawMessage m = new RawMessageImpl(id, headersAndPayload);
- ByteBuf serialized = m.serialize();
+ ByteBuf serialized = ByteBufPair.coalesce(m.serialize());
byte[] bytes = new byte[serialized.readableBytes()];
serialized.readBytes(bytes);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
index 9c465f746..ed3c1845a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
@@ -34,6 +34,7 @@
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.ClientConfiguration;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -107,6 +108,8 @@ public void initChannel(SocketChannel ch) throws Exception {
SslContext sslCtx = builder.build();
ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
}
+
+ ch.pipeline().addLast("ByteBufPairEncoder", ByteBufPair.ENCODER);
ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MaxMessageSize, 0, 4, 0, 4));
ch.pipeline().addLast("handler", new ClientCnx(conf, eventLoopGroup));
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index af9214302..112465802 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -47,7 +47,7 @@
import org.apache.pulsar.client.api.PulsarClientException.CryptoException;
import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.Commands.ChecksumType;
-import org.apache.pulsar.common.api.DoubleByteBuf;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.PulsarDecoder;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
@@ -98,7 +98,7 @@
private volatile long lastSequenceIdPublished;
private MessageCrypto msgCrypto = null;
-
+
private ScheduledFuture<?> keyGeneratorTask = null;
private final Map<String, String> metadata;
@@ -129,7 +129,7 @@ public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfiguration
if (conf.isEncryptionEnabled()) {
String logCtx = "[" + topic + "] [" + producerName + "] [" + producerId + "]";
- this.msgCrypto = new MessageCrypto(logCtx , true);
+ this.msgCrypto = new MessageCrypto(logCtx, true);
// Regenerate data key cipher at fixed interval
keyGeneratorTask = client.eventLoopGroup().scheduleWithFixedDelay(() -> {
@@ -261,7 +261,8 @@ public void sendAsync(Message message, SendCallback callback) {
if (compressedSize > PulsarDecoder.MaxMessageSize) {
compressedPayload.release();
String compressedStr = (!isBatchMessagingEnabled() && conf.getCompressionType() != CompressionType.NONE)
- ? "Compressed" : "";
+ ? "Compressed"
+ : "";
callback.sendComplete(new PulsarClientException.InvalidMessageException(
format("%s Message payload size %d cannot exceed %d bytes", compressedStr, compressedSize,
PulsarDecoder.MaxMessageSize)));
@@ -311,7 +312,7 @@ public void sendAsync(Message message, SendCallback callback) {
}
} else {
ByteBuf encryptedPayload = encryptMessage(msgMetadata, compressedPayload);
- ByteBuf cmd = sendMessage(producerId, sequenceId, 1, msgMetadata.build(), encryptedPayload);
+ ByteBufPair cmd = sendMessage(producerId, sequenceId, 1, msgMetadata.build(), encryptedPayload);
msgMetadata.recycle();
final OpSendMsg op = OpSendMsg.create(msg, cmd, sequenceId, callback);
@@ -350,14 +351,16 @@ public void sendAsync(Message message, SendCallback callback) {
}
}
- private ByteBuf encryptMessage(MessageMetadata.Builder msgMetadata, ByteBuf compressedPayload) throws PulsarClientException {
+ private ByteBuf encryptMessage(MessageMetadata.Builder msgMetadata, ByteBuf compressedPayload)
+ throws PulsarClientException {
ByteBuf encryptedPayload = compressedPayload;
if (!conf.isEncryptionEnabled() || msgCrypto == null) {
return encryptedPayload;
}
try {
- encryptedPayload = msgCrypto.encrypt(conf.getEncryptionKeys(), conf.getCryptoKeyReader(), msgMetadata, compressedPayload);
+ encryptedPayload = msgCrypto.encrypt(conf.getEncryptionKeys(), conf.getCryptoKeyReader(), msgMetadata,
+ compressedPayload);
} catch (PulsarClientException e) {
// Unless config is set to explicitly publish un-encrypted message upon failure, fail the request
if (conf.getCryptoFailureAction() == ProducerCryptoFailureAction.SEND) {
@@ -370,7 +373,7 @@ private ByteBuf encryptMessage(MessageMetadata.Builder msgMetadata, ByteBuf comp
return encryptedPayload;
}
- private ByteBuf sendMessage(long producerId, long sequenceId, int numMessages, MessageMetadata msgMetadata,
+ private ByteBufPair sendMessage(long producerId, long sequenceId, int numMessages, MessageMetadata msgMetadata,
ByteBuf compressedPayload) throws IOException {
ChecksumType checksumType;
@@ -696,11 +699,10 @@ protected synchronized void recoverChecksumError(ClientCnx cnx, long sequenceId)
* return false that means that message is corrupted. Returns true if checksum is not present.
*/
protected boolean verifyLocalBufferIsNotCorrupted(OpSendMsg op) {
- DoubleByteBuf msg = getDoubleByteBuf(op.cmd);
+ ByteBufPair msg = op.cmd;
if (msg != null) {
ByteBuf headerFrame = msg.getFirst();
- msg.markReaderIndex();
headerFrame.markReaderIndex();
try {
// skip bytes up to checksum index
@@ -720,11 +722,10 @@ protected boolean verifyLocalBufferIsNotCorrupted(OpSendMsg op) {
}
} finally {
headerFrame.resetReaderIndex();
- msg.resetReaderIndex();
}
return true;
} else {
- log.warn("[{}] Failed while casting {} into DoubleByteBuf", producerName, op.cmd.getClass().getName());
+ log.warn("[{}] Failed while casting {} into ByteBufPair", producerName, op.cmd.getClass().getName());
return false;
}
}
@@ -732,14 +733,14 @@ protected boolean verifyLocalBufferIsNotCorrupted(OpSendMsg op) {
protected static final class OpSendMsg {
MessageImpl msg;
List<MessageImpl> msgs;
- ByteBuf cmd;
+ ByteBufPair cmd;
SendCallback callback;
long sequenceId;
long createdAt;
long batchSizeByte = 0;
int numMessagesInBatch = 1;
- static OpSendMsg create(MessageImpl msg, ByteBuf cmd, long sequenceId, SendCallback callback) {
+ static OpSendMsg create(MessageImpl msg, ByteBufPair cmd, long sequenceId, SendCallback callback) {
OpSendMsg op = RECYCLER.get();
op.msg = msg;
op.cmd = cmd;
@@ -749,7 +750,7 @@ static OpSendMsg create(MessageImpl msg, ByteBuf cmd, long sequenceId, SendCallb
return op;
}
- static OpSendMsg create(List<MessageImpl> msgs, ByteBuf cmd, long sequenceId, SendCallback callback) {
+ static OpSendMsg create(List<MessageImpl> msgs, ByteBufPair cmd, long sequenceId, SendCallback callback) {
OpSendMsg op = RECYCLER.get();
op.msgs = msgs;
op.cmd = cmd;
@@ -812,9 +813,9 @@ void connectionOpened(final ClientCnx cnx) {
long requestId = client.newRequestId();
- cnx.sendRequestWithId(Commands.newProducer(topic, producerId, requestId, producerName,
- conf.isEncryptionEnabled(), metadata), requestId)
- .thenAccept(pair -> {
+ cnx.sendRequestWithId(
+ Commands.newProducer(topic, producerId, requestId, producerName, conf.isEncryptionEnabled(), metadata),
+ requestId).thenAccept(pair -> {
String producerName = pair.getLeft();
long lastSequenceId = pair.getRight();
@@ -971,12 +972,10 @@ private void resendMessages(ClientCnx cnx) {
* @param op
*/
private void stripChecksum(OpSendMsg op) {
- op.cmd.markReaderIndex();
int totalMsgBufSize = op.cmd.readableBytes();
- DoubleByteBuf msg = getDoubleByteBuf(op.cmd);
+ ByteBufPair msg = op.cmd;
if (msg != null) {
ByteBuf headerFrame = msg.getFirst();
- msg.markReaderIndex();
headerFrame.markReaderIndex();
try {
headerFrame.skipBytes(4); // skip [total-size]
@@ -986,7 +985,6 @@ private void stripChecksum(OpSendMsg op) {
headerFrame.skipBytes(cmdSize);
if (!hasChecksum(headerFrame)) {
- headerFrame.resetReaderIndex();
return;
}
@@ -1004,13 +1002,11 @@ private void stripChecksum(OpSendMsg op) {
headerFrame.writerIndex(headerSize); // set headerFrame write-index to overwrite metadata over checksum
metadata.readBytes(headerFrame, metadata.readableBytes());
headerFrame.capacity(headerFrameSize - checksumSize); // reduce capacity by removed checksum bytes
- headerFrame.resetReaderIndex();
-
} finally {
- op.cmd.resetReaderIndex();
+ headerFrame.resetReaderIndex();
}
} else {
- log.warn("[{}] Failed while casting {} into DoubleByteBuf", producerName, op.cmd.getClass().getName());
+ log.warn("[{}] Failed while casting {} into ByteBufPair", producerName, op.cmd.getClass().getName());
}
}
@@ -1157,7 +1153,7 @@ private void batchMessageAndSend() {
ByteBuf compressedPayload = batchMessageContainer.getCompressedBatchMetadataAndPayload();
long sequenceId = batchMessageContainer.sequenceId;
ByteBuf encryptedPayload = encryptMessage(batchMessageContainer.messageMetadata, compressedPayload);
- ByteBuf cmd = sendMessage(producerId, sequenceId, batchMessageContainer.numMessagesInBatch,
+ ByteBufPair cmd = sendMessage(producerId, sequenceId, batchMessageContainer.numMessagesInBatch,
batchMessageContainer.setBatchAndBuild(), encryptedPayload);
op = OpSendMsg.create(batchMessageContainer.messages, cmd, sequenceId,
@@ -1204,30 +1200,6 @@ private void batchMessageAndSend() {
}
}
- /**
- * Casts input cmd to {@link DoubleByteBuf}
- *
- * Incase if leak-detection level is enabled: pulsar instruments {@link DoubleByteBuf} into LeakAwareByteBuf (type
- * of {@link io.netty.buffer.WrappedByteBuf}) So, this method casts input cmd to {@link DoubleByteBuf} else
- * retrieves it from LeakAwareByteBuf.
- *
- * @param cmd
- * @return DoubleByteBuf or null in case failed to cast input {@link ByteBuf}
- */
- private DoubleByteBuf getDoubleByteBuf(ByteBuf cmd) {
- DoubleByteBuf msg = null;
- if (cmd instanceof DoubleByteBuf) {
- msg = (DoubleByteBuf) cmd;
- } else {
- try {
- msg = (DoubleByteBuf) cmd.unwrap();
- } catch (Exception e) {
- log.error("[{}] Failed while casting {} into DoubleByteBuf", producerName, cmd.getClass().getName(), e);
- }
- }
- return msg;
- }
-
public long getDelayInMillis() {
OpSendMsg firstMsg = pendingMessages.peek();
if (firstMsg != null) {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/ByteBufPair.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/ByteBufPair.java
new file mode 100644
index 000000000..bab0a5afb
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/ByteBufPair.java
@@ -0,0 +1,130 @@
+/**
+ * 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.pulsar.common.api;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.Recycler;
+import io.netty.util.Recycler.Handle;
+import io.netty.util.ReferenceCounted;
+
+/**
+ * ByteBuf holder that contains 2 buffers.
+ */
+public final class ByteBufPair extends AbstractReferenceCounted {
+
+ private ByteBuf b1;
+ private ByteBuf b2;
+ private final Handle<ByteBufPair> recyclerHandle;
+
+ private static final Recycler<ByteBufPair> RECYCLER = new Recycler<ByteBufPair>() {
+ @Override
+ protected ByteBufPair newObject(Recycler.Handle<ByteBufPair> handle) {
+ return new ByteBufPair(handle);
+ }
+ };
+
+ private ByteBufPair(Handle<ByteBufPair> recyclerHandle) {
+ this.recyclerHandle = recyclerHandle;
+ }
+
+ /**
+ * Get a new {@link ByteBufPair} from the pool and assign 2 buffers to it.
+ * <p>
+ * The buffers b1 and b2 lifecycles are now managed by the ByteBufPair: when the {@link ByteBufPair} is deallocated,
+ * b1 and b2 will be released as well.
+ *
+ * @param b1
+ * @param b2
+ * @return
+ */
+ public static ByteBufPair get(ByteBuf b1, ByteBuf b2) {
+ ByteBufPair buf = RECYCLER.get();
+ buf.setRefCnt(1);
+ buf.b1 = b1;
+ buf.b2 = b2;
+ return buf;
+ }
+
+ public ByteBuf getFirst() {
+ return b1;
+ }
+
+ public ByteBuf getSecond() {
+ return b2;
+ }
+
+ public int readableBytes() {
+ return b1.readableBytes() + b2.readableBytes();
+ }
+
+ /**
+ * @return a single buffer with the content of both individual buffers
+ */
+ @VisibleForTesting
+ public static ByteBuf coalesce(ByteBufPair pair) {
+ ByteBuf b = Unpooled.buffer(pair.readableBytes());
+ b.writeBytes(pair.b1, pair.b1.readerIndex(), pair.b1.readableBytes());
+ b.writeBytes(pair.b2, pair.b2.readerIndex(), pair.b2.readableBytes());
+ return b;
+ }
+
+ @Override
+ protected void deallocate() {
+ b1.release();
+ b2.release();
+ b1 = b2 = null;
+ recyclerHandle.recycle(this);
+ }
+
+ @Override
+ public ReferenceCounted touch(Object hint) {
+ b1.touch(hint);
+ b2.touch(hint);
+ return this;
+ }
+
+ public static final Encoder ENCODER = new Encoder();
+
+ @Sharable
+ public static class Encoder extends ChannelOutboundHandlerAdapter {
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+ if (msg instanceof ByteBufPair) {
+ ByteBufPair b = (ByteBufPair) msg;
+
+ // Write each buffer individually on the socket. The retain() here is needed to preserve the fact that
+ // ByteBuf are automatically released after a write. If the ByteBufPair ref count is increased and it
+ // gets written multiple times, the individual buffers refcount should be reflected as well.
+ ctx.write(b.getFirst().retain(), ctx.voidPromise());
+ ctx.write(b.getSecond().retain(), promise);
+ } else {
+ ctx.write(msg, promise);
+ }
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java
index 38fc36ed7..e7b670a63 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java
@@ -276,7 +276,7 @@ public static MessageMetadata parseMessageMetadata(ByteBuf buffer) {
}
}
- public static ByteBuf newMessage(long consumerId, MessageIdData messageId, ByteBuf metadataAndPayload) {
+ public static ByteBufPair newMessage(long consumerId, MessageIdData messageId, ByteBuf metadataAndPayload) {
CommandMessage.Builder msgBuilder = CommandMessage.newBuilder();
msgBuilder.setConsumerId(consumerId);
msgBuilder.setMessageId(messageId);
@@ -284,7 +284,7 @@ public static ByteBuf newMessage(long consumerId, MessageIdData messageId, ByteB
BaseCommand.Builder cmdBuilder = BaseCommand.newBuilder();
BaseCommand cmd = cmdBuilder.setType(Type.MESSAGE).setMessage(msg).build();
- ByteBuf res = serializeCommandMessageWithSize(cmd, metadataAndPayload);
+ ByteBufPair res = serializeCommandMessageWithSize(cmd, metadataAndPayload);
cmd.recycle();
cmdBuilder.recycle();
msg.recycle();
@@ -292,7 +292,7 @@ public static ByteBuf newMessage(long consumerId, MessageIdData messageId, ByteB
return res;
}
- public static ByteBuf newSend(long producerId, long sequenceId, int numMessages, ChecksumType checksumType,
+ public static ByteBufPair newSend(long producerId, long sequenceId, int numMessages, ChecksumType checksumType,
MessageMetadata messageData, ByteBuf payload) {
CommandSend.Builder sendBuilder = CommandSend.newBuilder();
sendBuilder.setProducerId(producerId);
@@ -302,7 +302,7 @@ public static ByteBuf newSend(long producerId, long sequenceId, int numMessages,
}
CommandSend send = sendBuilder.build();
- ByteBuf res = serializeCommandSendWithSize(BaseCommand.newBuilder().setType(Type.SEND).setSend(send),
+ ByteBufPair res = serializeCommandSendWithSize(BaseCommand.newBuilder().setType(Type.SEND).setSend(send),
checksumType, messageData, payload);
send.recycle();
sendBuilder.recycle();
@@ -674,8 +674,8 @@ private static ByteBuf serializeWithSize(BaseCommand.Builder cmdBuilder) {
return buf;
}
- private static ByteBuf serializeCommandSendWithSize(BaseCommand.Builder cmdBuilder, ChecksumType checksumType, MessageMetadata msgMetadata,
- ByteBuf payload) {
+ private static ByteBufPair serializeCommandSendWithSize(BaseCommand.Builder cmdBuilder, ChecksumType checksumType,
+ MessageMetadata msgMetadata, ByteBuf payload) {
// / Wire format
// [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD]
@@ -720,7 +720,7 @@ private static ByteBuf serializeCommandSendWithSize(BaseCommand.Builder cmdBuild
throw new RuntimeException(e);
}
- ByteBuf command = DoubleByteBuf.get(headers, payload);
+ ByteBufPair command = ByteBufPair.get(headers, payload);
// write checksum at created checksum-placeholder
if (includeChecksum) {
@@ -853,7 +853,7 @@ public static ByteBuf deSerializeSingleMessageInBatch(ByteBuf uncompressedPayloa
return singleMessagePayload;
}
- private static ByteBuf serializeCommandMessageWithSize(BaseCommand cmd, ByteBuf metadataAndPayload) {
+ private static ByteBufPair serializeCommandMessageWithSize(BaseCommand cmd, ByteBuf metadataAndPayload) {
// / Wire format
// [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD]
//
@@ -879,7 +879,7 @@ private static ByteBuf serializeCommandMessageWithSize(BaseCommand cmd, ByteBuf
throw new RuntimeException(e);
}
- return DoubleByteBuf.get(headers, metadataAndPayload);
+ return (ByteBufPair) ByteBufPair.get(headers, metadataAndPayload);
}
private static int getCurrentProtocolVersion() {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/DoubleByteBuf.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/DoubleByteBuf.java
deleted file mode 100644
index 66fd009cc..000000000
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/DoubleByteBuf.java
+++ /dev/null
@@ -1,485 +0,0 @@
-/**
- * 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.
- */
-/*
-* 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.pulsar.common.api;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.lang.reflect.Constructor;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-import java.nio.channels.FileChannel;
-import java.nio.channels.GatheringByteChannel;
-import java.nio.channels.ScatteringByteChannel;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.netty.buffer.AbstractReferenceCountedByteBuf;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.Unpooled;
-import io.netty.util.Recycler;
-import io.netty.util.Recycler.Handle;
-import io.netty.util.ResourceLeakDetector;
-import io.netty.util.ResourceLeakDetectorFactory;
-import io.netty.util.ResourceLeakTracker;
-
-/**
- * ByteBuf that holds 2 buffers. Similar to {@see CompositeByteBuf} but doesn't allocate list to hold them.
- */
-@SuppressWarnings("unchecked")
-public final class DoubleByteBuf extends AbstractReferenceCountedByteBuf {
-
- private ByteBuf b1;
- private ByteBuf b2;
- private final Handle<DoubleByteBuf> recyclerHandle;
-
- private static final Recycler<DoubleByteBuf> RECYCLER = new Recycler<DoubleByteBuf>() {
- @Override
- protected DoubleByteBuf newObject(Recycler.Handle<DoubleByteBuf> handle) {
- return new DoubleByteBuf(handle);
- }
- };
-
- private DoubleByteBuf(Handle<DoubleByteBuf> recyclerHandle) {
- super(Integer.MAX_VALUE);
- this.recyclerHandle = recyclerHandle;
- }
-
- public static ByteBuf get(ByteBuf b1, ByteBuf b2) {
- DoubleByteBuf buf = RECYCLER.get();
- buf.setRefCnt(1);
-
- // Make sure the buffers are not deallocated as long as we hold them. Also, buffers can get retained/releases
- // outside of DoubleByteBuf scope
- buf.b1 = b1.retain();
- buf.b2 = b2.retain();
- buf.setIndex(0, b1.readableBytes() + b2.readableBytes());
- return toLeakAwareBuffer(buf);
- }
-
- public ByteBuf getFirst() {
- return b1;
- }
-
- public ByteBuf getSecond() {
- return b2;
- }
-
- @Override
- public boolean isDirect() {
- return b1.isDirect() && b2.isDirect();
- }
-
- @Override
- public boolean hasArray() {
- // There's no single array available
- return false;
- }
-
- @Override
- public byte[] array() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public int arrayOffset() {
-
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean hasMemoryAddress() {
- return false;
- }
-
- @Override
- public long memoryAddress() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public int capacity() {
- return b1.writerIndex() + b2.writerIndex();
- }
-
- @Override
- public int maxCapacity() {
- return capacity();
- }
-
- @Override
- public int writableBytes() {
- return 0;
- }
-
- @Override
- public DoubleByteBuf capacity(int newCapacity) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public ByteBufAllocator alloc() {
- return PooledByteBufAllocator.DEFAULT;
- }
-
- @Override
- public ByteOrder order() {
- return ByteOrder.BIG_ENDIAN;
- }
-
- @Override
- public byte getByte(int index) {
- if (index < b1.writerIndex()) {
- return b1.getByte(index);
- } else {
- return b2.getByte(index - b1.writerIndex());
- }
- }
-
- @Override
- protected byte _getByte(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected short _getShort(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected int _getUnsignedMedium(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected int _getInt(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected long _getLong(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf getBytes(int index, byte[] dst, int dstIndex, int length) {
- return getBytes(index, Unpooled.wrappedBuffer(dst), dstIndex, length);
- }
-
- @Override
- public ByteBuf getBytes(int index, ByteBuffer dst) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf getBytes(int index, ByteBuf dst, int dstIndex, int length) {
- checkDstIndex(index, length, dstIndex, dst.capacity());
- if (length == 0) {
- return this;
- }
-
- int b1Length = Math.min(length, b1.readableBytes() - index);
- if (b1Length > 0) {
- b1.getBytes(b1.readerIndex() + index, dst, dstIndex, b1Length);
- dstIndex += b1Length;
- length -= b1Length;
- index = 0;
- } else {
- index -= b1.readableBytes();
- }
-
- if (length > 0) {
- int b2Length = Math.min(length, b2.readableBytes() - index);
- b2.getBytes(b2.readerIndex() + index, dst, dstIndex, b2Length);
- }
- return this;
- }
-
- @Override
- public int getBytes(int index, GatheringByteChannel out, int length) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf getBytes(int index, OutputStream out, int length) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf setByte(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected void _setByte(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf setShort(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected void _setShort(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf setMedium(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected void _setMedium(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf setInt(int index, int value) {
- return (DoubleByteBuf) super.setInt(index, value);
- }
-
- @Override
- protected void _setInt(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf setLong(int index, long value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected void _setLong(int index, long value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf setBytes(int index, byte[] src, int srcIndex, int length) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf setBytes(int index, ByteBuffer src) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public DoubleByteBuf setBytes(int index, ByteBuf src, int srcIndex, int length) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public int setBytes(int index, InputStream in, int length) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public int setBytes(int index, ScatteringByteChannel in, int length) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected short _getShortLE(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected int _getUnsignedMediumLE(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected int _getIntLE(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected long _getLongLE(int index) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected void _setShortLE(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected void _setMediumLE(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected void _setIntLE(int index, int value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- protected void _setLongLE(int index, long value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public int getBytes(int index, FileChannel out, long position, int length) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public int setBytes(int index, FileChannel in, long position, int length) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public ByteBuf copy(int index, int length) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public int nioBufferCount() {
- return b1.nioBufferCount() + b2.nioBufferCount();
- }
-
- @Override
- public ByteBuffer internalNioBuffer(int index, int length) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public ByteBuffer nioBuffer(int index, int length) {
- ByteBuffer dst = ByteBuffer.allocate(length);
- ByteBuf b = Unpooled.wrappedBuffer(dst);
- b.writerIndex(0);
- getBytes(index, b, length);
- return dst;
- }
-
- @Override
- public ByteBuffer[] nioBuffers(int index, int length) {
- return new ByteBuffer[] { nioBuffer(index, length) };
- }
-
- @Override
- public DoubleByteBuf discardReadBytes() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public String toString() {
- String result = super.toString();
- result = result.substring(0, result.length() - 1);
- return result + ", components=2)";
- }
-
- @Override
- public ByteBuffer[] nioBuffers() {
- return nioBuffers(readerIndex(), readableBytes());
- }
-
- @Override
- protected void deallocate() {
- // Double release of buffer for the initial ref-count and the internal retain() when the DoubleByteBuf was
- // created
- b1.release(2);
- b2.release(2);
- b1 = b2 = null;
- recyclerHandle.recycle(this);
- }
-
- @Override
- public ByteBuf unwrap() {
- return null;
- }
-
- private static final Logger log = LoggerFactory.getLogger(DoubleByteBuf.class);
-
- private static final ResourceLeakDetector<DoubleByteBuf> leakDetector = ResourceLeakDetectorFactory.instance()
- .newResourceLeakDetector(DoubleByteBuf.class);
- private static final Constructor<ByteBuf> simpleLeakAwareByteBufConstructor;
- private static final Constructor<ByteBuf> advancedLeakAwareByteBufConstructor;
-
- static {
- Constructor<ByteBuf> _simpleLeakAwareByteBufConstructor = null;
- Constructor<ByteBuf> _advancedLeakAwareByteBufConstructor = null;
- try {
- Class<?> simpleLeakAwareByteBufClass = Class.forName("io.netty.buffer.SimpleLeakAwareByteBuf");
- _simpleLeakAwareByteBufConstructor = (Constructor<ByteBuf>) simpleLeakAwareByteBufClass
- .getDeclaredConstructor(ByteBuf.class, ResourceLeakTracker.class);
- _simpleLeakAwareByteBufConstructor.setAccessible(true);
-
- Class<?> advancedLeakAwareByteBufClass = Class.forName("io.netty.buffer.AdvancedLeakAwareByteBuf");
- _advancedLeakAwareByteBufConstructor = (Constructor<ByteBuf>) advancedLeakAwareByteBufClass
- .getDeclaredConstructor(ByteBuf.class, ResourceLeakTracker.class);
- _advancedLeakAwareByteBufConstructor.setAccessible(true);
- } catch (Exception e) {
- log.error("Failed to use reflection to enable leak detection");
- } finally {
- simpleLeakAwareByteBufConstructor = _simpleLeakAwareByteBufConstructor;
- advancedLeakAwareByteBufConstructor = _advancedLeakAwareByteBufConstructor;
- }
- }
-
- private static ByteBuf toLeakAwareBuffer(DoubleByteBuf buf) {
- try {
- ResourceLeakTracker<DoubleByteBuf> leak;
- switch (ResourceLeakDetector.getLevel()) {
- case DISABLED:
- break;
-
- case SIMPLE:
- leak = leakDetector.track(buf);
- if (leak != null) {
- return simpleLeakAwareByteBufConstructor.newInstance(buf, leak);
- }
- break;
- case ADVANCED:
- case PARANOID:
- leak = leakDetector.track(buf);
- if (leak != null) {
- return advancedLeakAwareByteBufConstructor.newInstance(buf, leak);
- }
- break;
- }
- return buf;
- } catch (Throwable t) {
- // Catch reflection exception
- throw new RuntimeException(t);
- }
- }
-}
\ No newline at end of file
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/api/DoubleByteBufTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/api/ByteBufPairTest.java
similarity index 54%
rename from pulsar-common/src/test/java/org/apache/pulsar/common/api/DoubleByteBufTest.java
rename to pulsar-common/src/test/java/org/apache/pulsar/common/api/ByteBufPairTest.java
index 29c7bad67..cff92c8cc 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/api/DoubleByteBufTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/api/ByteBufPairTest.java
@@ -20,36 +20,28 @@
import static org.testng.Assert.assertEquals;
-import org.apache.pulsar.common.api.DoubleByteBuf;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.testng.annotations.Test;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
-public class DoubleByteBufTest {
+public class ByteBufPairTest {
- /**
- * Verify that readableBytes() returns writerIndex - readerIndex. In this case writerIndex is the end of the buffer
- * and readerIndex is increased by 64.
- *
- * @throws Exception
- */
@Test
- public void testReadableBytes() throws Exception {
-
+ public void testDoubleByteBuf() throws Exception {
ByteBuf b1 = PooledByteBufAllocator.DEFAULT.heapBuffer(128, 128);
b1.writerIndex(b1.capacity());
ByteBuf b2 = PooledByteBufAllocator.DEFAULT.heapBuffer(128, 128);
b2.writerIndex(b2.capacity());
- ByteBuf buf = DoubleByteBuf.get(b1, b2);
+ ByteBufPair buf = ByteBufPair.get(b1, b2);
- assertEquals(buf.readerIndex(), 0);
- assertEquals(buf.writerIndex(), 256);
assertEquals(buf.readableBytes(), 256);
+ assertEquals(buf.getFirst(), b1);
+ assertEquals(buf.getSecond(), b2);
- for (int i = 0; i < 4; ++i) {
- buf.skipBytes(64);
- assertEquals(buf.readableBytes(), 256 - 64 * (i + 1));
- }
+ assertEquals(buf.refCnt(), 1);
+ assertEquals(b1.refCnt(), 1);
+ assertEquals(b2.refCnt(), 1);
buf.release();
@@ -58,22 +50,4 @@ public void testReadableBytes() throws Exception {
assertEquals(b2.refCnt(), 0);
}
- @Test
- public void testCapacity() throws Exception {
-
- ByteBuf b1 = PooledByteBufAllocator.DEFAULT.heapBuffer(128, 128);
- b1.writerIndex(b1.capacity());
- ByteBuf b2 = PooledByteBufAllocator.DEFAULT.heapBuffer(128, 128);
- b2.writerIndex(b2.capacity());
- ByteBuf buf = DoubleByteBuf.get(b1, b2);
-
- assertEquals(buf.capacity(), 256);
- assertEquals(buf.maxCapacity(), 256);
-
- buf.release();
-
- assertEquals(buf.refCnt(), 0);
- assertEquals(b1.refCnt(), 0);
- assertEquals(b2.refCnt(), 0);
- }
}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/compression/CommandsTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/compression/CommandsTest.java
index aefdcae45..808552fe1 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/compression/CommandsTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/compression/CommandsTest.java
@@ -22,12 +22,11 @@
import static org.testng.Assert.assertTrue;
import java.io.IOException;
-import java.nio.ByteBuffer;
import org.apache.pulsar.checksum.utils.Crc32cChecksum;
import org.apache.pulsar.common.api.Commands;
-import org.apache.pulsar.common.api.DoubleByteBuf;
import org.apache.pulsar.common.api.Commands.ChecksumType;
+import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream;
import org.testng.annotations.Test;
@@ -48,20 +47,19 @@ public void testChecksumSendCommand() throws Exception {
MessageMetadata messageMetadata = MessageMetadata.newBuilder().setPublishTime(System.currentTimeMillis())
.setProducerName(producerName).setSequenceId(sequenceId).build();
int expectedChecksum = computeChecksum(messageMetadata, data);
- ByteBuf clientCommand = Commands.newSend(1, 0, 1, ChecksumType.Crc32c, messageMetadata, data);
+ ByteBufPair clientCommand = Commands.newSend(1, 0, 1, ChecksumType.Crc32c, messageMetadata, data);
clientCommand.retain();
- ByteBuffer inputBytes = clientCommand.nioBuffer();
- ByteBuf receivedBuf = Unpooled.wrappedBuffer(inputBytes);
+ ByteBuf receivedBuf = ByteBufPair.coalesce(clientCommand);
receivedBuf.skipBytes(4); //skip [total-size]
int cmdSize = (int) receivedBuf.readUnsignedInt();
receivedBuf.readerIndex(8 + cmdSize);
int startMessagePos = receivedBuf.readerIndex();
-
+
/*** 1. verify checksum and metadataParsing ***/
boolean hasChecksum = Commands.hasChecksum(receivedBuf);
int checksum = Commands.readChecksum(receivedBuf).intValue();
-
-
+
+
// verify checksum is present
assertTrue(hasChecksum);
// verify checksum value
@@ -69,13 +67,13 @@ public void testChecksumSendCommand() throws Exception {
MessageMetadata metadata = Commands.parseMessageMetadata(receivedBuf);
// verify metadata parsing
assertEquals(metadata.getProducerName(), producerName);
-
- /** 2. parseMessageMetadata should skip checksum if present **/
+
+ /** 2. parseMessageMetadata should skip checksum if present **/
receivedBuf.readerIndex(startMessagePos);
metadata = Commands.parseMessageMetadata(receivedBuf);
// verify metadata parsing
assertEquals(metadata.getProducerName(), producerName);
-
+
}
private int computeChecksum(MessageMetadata msgMetadata, ByteBuf compressedPayload) throws IOException {
@@ -86,12 +84,12 @@ private int computeChecksum(MessageMetadata msgMetadata, ByteBuf compressedPaylo
metaPayloadFrame.writeInt(metadataSize);
msgMetadata.writeTo(outStream);
ByteBuf payload = compressedPayload.copy();
- ByteBuf metaPayloadBuf = DoubleByteBuf.get(metaPayloadFrame, payload);
- int computedChecksum = Crc32cChecksum.computeChecksum(metaPayloadBuf);
+ ByteBufPair metaPayloadBuf = ByteBufPair.get(metaPayloadFrame, payload);
+ int computedChecksum = Crc32cChecksum.computeChecksum(ByteBufPair.coalesce(metaPayloadBuf));
outStream.recycle();
metaPayloadBuf.release();
return computedChecksum;
}
-
+
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services