Skip to content

Commit 1397faf

Browse files
dlg99dlg99
committed
removed usage of byte[] where possible
1 parent 8db72f4 commit 1397faf

File tree

2 files changed

+50
-40
lines changed

2 files changed

+50
-40
lines changed

‎managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

+47-37
Original file line numberDiff line numberDiff line change
@@ -39,10 +39,8 @@
3939
import io.netty.buffer.ByteBufUtil;
4040
import io.netty.buffer.CompositeByteBuf;
4141
import io.netty.buffer.Unpooled;
42-
import java.io.ByteArrayInputStream;
43-
import java.io.ByteArrayOutputStream;
44-
import java.io.DataInputStream;
4542
import java.io.IOException;
43+
import java.nio.charset.StandardCharsets;
4644
import java.time.Clock;
4745
import java.util.ArrayDeque;
4846
import java.util.ArrayList;
@@ -636,7 +634,7 @@ private void recoverFromLedgerByEntryId(ManagedCursorInfo info,
636634
}
637635

638636
LedgerEntry entry = seq.nextElement();
639-
byte[] data = entry.getEntry();
637+
ByteBuf data = entry.getEntryBuffer();
640638
try {
641639
ChunkSequenceFooter chunkSequenceFooter = parseChunkSequenceFooter(data);
642640
if (chunkSequenceFooter.numParts > 0) {
@@ -672,23 +670,28 @@ private void readChunkSequence(VoidCallback callback, LedgerHandle lh,
672670
lh.asyncReadEntries(startPos, endPos, new AsyncCallback.ReadCallback() {
673671
@Override
674672
public void readComplete(int rc, LedgerHandle lh, Enumeration<LedgerEntry> entries, Object ctx) {
675-
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
673+
CompositeByteBuf buffer = PulsarByteBufAllocator.DEFAULT.compositeBuffer();
674+
675+
AtomicInteger readableBytes = new AtomicInteger(0);
676676
entries.asIterator().forEachRemaining(entry -> {
677-
log.info("pos {} len {} bytes ", entry.getEntryId(), entry.getLength());
678-
try {
679-
buffer.write(entry.getEntry());
680-
} catch (IOException err) {
681-
throw new RuntimeException(err);
677+
if (log.isInfoEnabled()) {
678+
log.debug("pos {} len {} bytes ", entry.getEntryId(), entry.getLength());
682679
}
680+
ByteBuf part = entry.getEntryBuffer();
681+
buffer.addComponent(part);
682+
readableBytes.addAndGet(part.readableBytes());
683683
});
684-
byte[] result = buffer.toByteArray();
684+
buffer.writerIndex(readableBytes.get())
685+
.readerIndex(0);
686+
685687
log.info("Read {} chunks, total of {} bytes, expected {} bytes", chunkSequenceFooter.numParts,
686-
result.length, chunkSequenceFooter.length);
687-
if (result.length != chunkSequenceFooter.length) {
688+
buffer.readableBytes(), chunkSequenceFooter.length);
689+
if (buffer.readableBytes() != chunkSequenceFooter.length) {
688690
callback.operationFailed(ManagedLedgerException.getManagedLedgerException(new IOException(
689-
"Expected " + chunkSequenceFooter.length + " bytes but read " + result.length + " bytes")));
691+
"Expected " + chunkSequenceFooter.length + " bytes but read "
692+
+ buffer.readableBytes() + " bytes")));
690693
}
691-
Throwable res = tryCompleteCursorRecovery(lh, result);
694+
Throwable res = tryCompleteCursorRecovery(lh, buffer);
692695
if (res == null) {
693696
callback.operationComplete();
694697
} else {
@@ -709,32 +712,42 @@ public static final class ChunkSequenceFooter {
709712
private int length;
710713
}
711714

712-
private ChunkSequenceFooter parseChunkSequenceFooter(byte[] data) throws IOException {
713-
if (data.length == 0 || data[0] != '{') {
715+
private ChunkSequenceFooter parseChunkSequenceFooter(ByteBuf data) throws IOException {
716+
// getChar() doesn't move the reader index
717+
if (data.readableBytes() == 0 || data.getByte(0) != '{') {
714718
// this is not JSON
715719
return ChunkSequenceFooter.NOT_CHUNKED;
716720
}
717-
return ObjectMapperFactory.getMapper().getObjectMapper().readValue(data, ChunkSequenceFooter.class);
721+
722+
try {
723+
return ObjectMapperFactory.getMapper().getObjectMapper()
724+
.readValue(data.toString(StandardCharsets.UTF_8), ChunkSequenceFooter.class);
725+
} catch (JsonProcessingException e) {
726+
return ChunkSequenceFooter.NOT_CHUNKED;
727+
}
718728
}
719729

720-
private Throwable tryCompleteCursorRecovery(LedgerHandle lh, byte[] data) {
721-
mbean.addReadCursorLedgerSize(data.length);
730+
private Throwable tryCompleteCursorRecovery(LedgerHandle lh, ByteBuf data) {
731+
mbean.addReadCursorLedgerSize(data.readableBytes());
722732

723733
try {
724734
data = decompressDataIfNeeded(data, lh);
725735
} catch (Throwable e) {
736+
data.release();
726737
log.error("[{}] Failed to decompress position info from ledger {} for cursor {}: {}", ledger.getName(),
727738
lh.getId(), name, e);
728739
return e;
729740
}
730741

731742
PositionInfo positionInfo;
732743
try {
733-
positionInfo = PositionInfo.parseFrom(data);
744+
positionInfo = PositionInfo.parseFrom(data.nioBuffer());
734745
} catch (InvalidProtocolBufferException e) {
735746
log.error("[{}] Failed to parse position info from ledger {} for cursor {}: {}", ledger.getName(),
736747
lh.getId(), name, e);
737748
return e;
749+
} finally {
750+
data.release();
738751
}
739752

740753
Map<String, Long> recoveredProperties = Collections.emptyMap();
@@ -3492,42 +3505,39 @@ private ByteBuf compressDataIfNeeded(ByteBuf data, LedgerHandle lh) {
34923505
result.readerIndex(0)
34933506
.writerIndex(4 + compressedSize);
34943507

3495-
int ratio = (int) (compressedSize * 100.0 / uncompressedSize);
3496-
log.info("[{}] Cursor {} Compressed data size {} bytes (with {}, original size {} bytes, ratio {}%)",
3497-
ledger.getName(), name, compressedSize, pulsarCursorInfoCompressionString, uncompressedSize, ratio);
3508+
if (log.isInfoEnabled()) {
3509+
int ratio = (int) (compressedSize * 100.0 / uncompressedSize);
3510+
log.info("[{}] Cursor {} Compressed data size {} bytes (with {}, original size {} bytes, ratio {}%)",
3511+
ledger.getName(), name, compressedSize, pulsarCursorInfoCompressionString,
3512+
uncompressedSize, ratio);
3513+
}
34983514
return result;
34993515
} finally {
35003516
data.release();
35013517
}
35023518
}
35033519

3504-
static byte[] decompressDataIfNeeded(byte[] data, LedgerHandle lh) {
3520+
static ByteBuf decompressDataIfNeeded(ByteBuf data, LedgerHandle lh) {
35053521
byte[] pulsarCursorInfoCompression =
35063522
lh.getCustomMetadata().get(METADATA_PROPERTY_CURSOR_COMPRESSION_TYPE);
35073523
if (pulsarCursorInfoCompression != null) {
35083524
String pulsarCursorInfoCompressionString = new String(pulsarCursorInfoCompression);
35093525
if (log.isDebugEnabled()) {
35103526
log.debug("Ledger {} compression {} decompressing {} bytes, full {}",
3511-
lh.getId(), pulsarCursorInfoCompressionString, data.length,
3527+
lh.getId(), pulsarCursorInfoCompressionString, data.readableBytes(),
35123528
ByteBufUtil.prettyHexDump(Unpooled.wrappedBuffer(data)));
35133529
}
3514-
ByteArrayInputStream input = new ByteArrayInputStream(data);
3515-
DataInputStream dataInputStream = new DataInputStream(input);
35163530
try {
3517-
int uncompressedSize = dataInputStream.readInt();
3518-
byte[] compressedData = dataInputStream.readAllBytes();
3531+
// this moves readerIndex
3532+
int uncompressedSize = data.readInt();
35193533
CompressionCodec compressionCodec = CompressionCodecProvider.getCompressionCodec(
35203534
CompressionType.valueOf(pulsarCursorInfoCompressionString));
3521-
ByteBuf decode = compressionCodec.decode(Unpooled.wrappedBuffer(compressedData), uncompressedSize);
3522-
try {
3523-
return ByteBufUtil.getBytes(decode);
3524-
} finally {
3525-
decode.release();
3526-
}
3535+
ByteBuf decode = compressionCodec.decode(data, uncompressedSize);
3536+
return decode;
35273537
} catch (IOException | MalformedInputException error) {
35283538
log.error("Cannot decompress cursor position using {}. Payload is {}",
35293539
pulsarCursorInfoCompressionString,
3530-
ByteBufUtil.prettyHexDump(Unpooled.wrappedBuffer(data)), error);
3540+
ByteBufUtil.prettyHexDump(data), error);
35313541
throw new RuntimeException(error);
35323542
}
35333543
} else {

‎managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,7 @@ public void testCloseCursor() throws Exception {
149149
ledger.addEntry(new byte[]{3});
150150
ledger.addEntry(new byte[]{4});
151151
ledger.addEntry(new byte[]{5});
152-
// Persistent cursor info to ledger.
152+
// Persist cursor info to ledger.
153153
c1.delete(PositionFactory.create(c1.getReadPosition().getLedgerId(), c1.getReadPosition().getEntryId()));
154154
Awaitility.await().until(() ->c1.getStats().getPersistLedgerSucceed() > 0);
155155
// Make cursor ledger can not work.
@@ -3276,9 +3276,9 @@ public void operationFailed(MetaStoreException e) {
32763276
try {
32773277
LedgerEntry entry = seq.nextElement();
32783278
PositionInfo positionInfo;
3279-
byte[] data = entry.getEntry();
3279+
ByteBuf data = entry.getEntryBuffer();
32803280
data = ManagedCursorImpl.decompressDataIfNeeded(data, lh);
3281-
positionInfo = PositionInfo.parseFrom(data);
3281+
positionInfo = PositionInfo.parseFrom(data.nioBuffer());
32823282
individualDeletedMessagesCount.set(positionInfo.getIndividualDeletedMessagesCount());
32833283
} catch (Exception e) {
32843284
}

0 commit comments

Comments
 (0)