Skip to content
This repository was archived by the owner on Jan 24, 2024. It is now read-only.

Upgrade pulsar with LightProto support #337

Merged
jiazhai merged 8 commits into from
Jan 26, 2021
Merged

Upgrade pulsar with LightProto support #337

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
20db368
Upgrade pulsar with LightProto support
BewareMyPower Jan 20, 2021
1e9601b
Fix PulsarEntryFormatter
BewareMyPower Jan 21, 2021
c8248c1
Fix PulsarEntryFormatter#decode
BewareMyPower Jan 21, 2021
06f8438
Fix exception for message without key
BewareMyPower Jan 21, 2021
141c4c5
Bump pulsar version to 2.8.0-SNAPSHOT
BewareMyPower Jan 21, 2021
9d893b4
Fix tests setup failure
BewareMyPower Jan 25, 2021
6756989
Bump pulsar to rc version
BewareMyPower Jan 26, 2021
3ef7fbc
Fix timestamp related tests
BewareMyPower Jan 26, 2021
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
4 changes: 2 additions & 2 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/InternalProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
import org.apache.pulsar.broker.service.Producer;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.api.proto.ProducerAccessMode;

/**
* InternalServerCnx, this only used to construct internalProducer / internalConsumer.
Expand All @@ -34,7 +34,7 @@ public InternalProducer(Topic topic, ServerCnx cnx,
long producerId, String producerName) {
super(topic, cnx, producerId, producerName, null,
false, null, null, 0, false,
PulsarApi.ProducerAccessMode.Shared, Optional.empty());
ProducerAccessMode.Shared, Optional.empty());
this.serverCnx = cnx;
}

Expand Down
17 changes: 8 additions & 9 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,8 +136,8 @@
import org.apache.pulsar.broker.loadbalance.LoadManager;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.api.proto.PulsarMarkers;
import org.apache.pulsar.common.api.proto.MarkerType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
Expand Down Expand Up @@ -1438,26 +1438,25 @@ private CompletableFuture<Long> writeTxnMarker(TopicPartition topicPartition,

private ByteBuf generateTxnMarker(TransactionResult transactionResult, long producerId, short producerEpoch) {
ControlRecordType controlRecordType;
PulsarMarkers.MarkerType markerType;
MarkerType markerType;
if (transactionResult.equals(TransactionResult.COMMIT)) {
markerType = PulsarMarkers.MarkerType.TXN_COMMIT;
markerType = MarkerType.TXN_COMMIT;
controlRecordType = ControlRecordType.COMMIT;
} else {
markerType = PulsarMarkers.MarkerType.TXN_ABORT;
markerType = MarkerType.TXN_ABORT;
controlRecordType = ControlRecordType.ABORT;
}
EndTransactionMarker marker = new EndTransactionMarker(controlRecordType, 0);
MemoryRecords memoryRecords = MemoryRecords.withEndTransactionMarker(producerId, producerEpoch, marker);

ByteBuf byteBuf = Unpooled.wrappedBuffer(memoryRecords.buffer());
PulsarApi.MessageMetadata messageMetadata = PulsarApi.MessageMetadata.newBuilder()
MessageMetadata messageMetadata = new MessageMetadata()
.setTxnidMostBits(producerId)
.setTxnidLeastBits(producerEpoch)
.setMarkerType(markerType.getNumber())
.setMarkerType(markerType.getValue())
.setPublishTime(SystemTime.SYSTEM.milliseconds())
.setProducerName("")
.setSequenceId(0L)
.build();
.setSequenceId(0L);
return Commands.serializeMetadataAndPayload(Commands.ChecksumType.None, messageMetadata, byteBuf);
}

Expand Down
2 changes: 1 addition & 1 deletion kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/MessageFetchContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -455,7 +455,7 @@ public void readEntriesFailed(ManagedLedgerException e, Object o) {

readFuture.completeExceptionally(e);
}
}, null);
}, null, PositionImpl.latest);

readFutures.putIfAbsent(cursorOffsetPair.getKey(), readFuture);
});
Expand Down
21 changes: 10 additions & 11 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/KafkaEntryFormatter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
import org.apache.kafka.common.record.MemoryRecordsBuilder;
import org.apache.kafka.common.record.Record;
import org.apache.kafka.common.record.TimestampType;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.Commands;

/**
Expand Down Expand Up @@ -73,17 +73,16 @@ public MemoryRecords decode(List<Entry> entries, byte magic) {
return builder.build();
}

private static PulsarApi.MessageMetadata getMessageMetadataWithNumberMessages(int numMessages) {
final PulsarApi.MessageMetadata.Builder builder = PulsarApi.MessageMetadata.newBuilder();
builder.addProperties(PulsarApi.KeyValue.newBuilder()
private static MessageMetadata getMessageMetadataWithNumberMessages(int numMessages) {
final MessageMetadata metadata = new MessageMetadata();
metadata.addProperty()
.setKey("entry.format")
.setValue(EntryFormatterFactory.EntryFormat.KAFKA.name().toLowerCase())
.build());
builder.setProducerName("");
builder.setSequenceId(0L);
builder.setPublishTime(System.currentTimeMillis());
builder.setNumMessagesInBatch(numMessages);
return builder.build();
.setValue(EntryFormatterFactory.EntryFormat.KAFKA.name().toLowerCase());
metadata.setProducerName("");
metadata.setSequenceId(0L);
metadata.setPublishTime(System.currentTimeMillis());
metadata.setNumMessagesInBatch(numMessages);
return metadata;
}

}
64 changes: 24 additions & 40 deletions ...a-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/PulsarEntryFormatter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,11 +41,10 @@
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.api.proto.PulsarApi.KeyValue;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.api.proto.PulsarApi.SingleMessageMetadata;
import org.apache.pulsar.common.api.proto.PulsarMarkers;
import org.apache.pulsar.common.api.proto.KeyValue;
import org.apache.pulsar.common.api.proto.MarkerType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.api.proto.SingleMessageMetadata;
import org.apache.pulsar.common.compression.CompressionCodec;
import org.apache.pulsar.common.compression.CompressionCodecProvider;
import org.apache.pulsar.common.protocol.Commands;
Expand All @@ -69,61 +68,45 @@ public ByteBuf encode(final MemoryRecords records, final int numMessages) {
int numMessagesInBatch = 0;

long sequenceId = -1;
// TODO: handle different compression type
PulsarApi.CompressionType compressionType = PulsarApi.CompressionType.NONE;

ByteBuf batchedMessageMetadataAndPayload = PulsarByteBufAllocator.DEFAULT
.buffer(Math.min(INITIAL_BATCH_BUFFER_SIZE, MAX_MESSAGE_BATCH_SIZE_BYTES));
List<MessageImpl<byte[]>> messages = Lists.newArrayListWithExpectedSize(numMessages);
MessageMetadata.Builder messageMetaBuilder = MessageMetadata.newBuilder();
final MessageMetadata msgMetadata = new MessageMetadata();

records.batches().forEach(recordBatch -> {
StreamSupport.stream(recordBatch.spliterator(), true).forEachOrdered(record -> {
MessageImpl<byte[]> message = recordToEntry(record);
messages.add(message);
if (messageMetaBuilder.getPublishTime() <= 0) {
messageMetaBuilder.setPublishTime(message.getPublishTime());
}
if (recordBatch.isTransactional()) {
messageMetaBuilder.setTxnidMostBits(recordBatch.producerId());
messageMetaBuilder.setTxnidLeastBits(recordBatch.producerEpoch());
msgMetadata.setTxnidMostBits(recordBatch.producerId());
msgMetadata.setTxnidLeastBits(recordBatch.producerEpoch());
}
});
});

for (MessageImpl<byte[]> message : messages) {
if (++numMessagesInBatch == 1) {
sequenceId = Commands.initBatchMessageMetadata(messageMetaBuilder, message.getMessageBuilder());
// msgMetadata will set publish time here
sequenceId = Commands.initBatchMessageMetadata(msgMetadata, message.getMessageBuilder());
}
currentBatchSizeBytes += message.getDataBuffer().readableBytes();
if (log.isDebugEnabled()) {
log.debug("recordsToByteBuf , sequenceId: {}, numMessagesInBatch: {}, currentBatchSizeBytes: {} ",
sequenceId, numMessagesInBatch, currentBatchSizeBytes);
}

PulsarApi.MessageMetadata.Builder msgBuilder = message.getMessageBuilder();
final MessageMetadata msgBuilder = message.getMessageBuilder();
batchedMessageMetadataAndPayload = Commands.serializeSingleMessageInBatchWithPayload(msgBuilder,
message.getDataBuffer(), batchedMessageMetadataAndPayload);
msgBuilder.recycle();
}

int uncompressedSize = batchedMessageMetadataAndPayload.readableBytes();

if (PulsarApi.CompressionType.NONE != compressionType) {
messageMetaBuilder.setCompression(compressionType);
messageMetaBuilder.setUncompressedSize(uncompressedSize);
}

messageMetaBuilder.setNumMessagesInBatch(numMessagesInBatch);

MessageMetadata msgMetadata = messageMetaBuilder.build();
msgMetadata.setNumMessagesInBatch(numMessagesInBatch);

ByteBuf buf = Commands.serializeMetadataAndPayload(ChecksumType.Crc32c,
msgMetadata,
batchedMessageMetadataAndPayload);

messageMetaBuilder.recycle();
msgMetadata.recycle();
batchedMessageMetadataAndPayload.release();

return buf;
Expand All @@ -138,19 +121,20 @@ public MemoryRecords decode(final List<Entry> entries, final byte magic) {
// each entry is a batched message
ByteBuf metadataAndPayload = entry.getDataBuffer();

// Uncompress the payload if necessary
Commands.skipBrokerEntryMetadataIfExist(metadataAndPayload);
MessageMetadata msgMetadata = Commands.parseMessageMetadata(metadataAndPayload);

if (msgMetadata.getMarkerType() == PulsarMarkers.MarkerType.TXN_COMMIT_VALUE
|| msgMetadata.getMarkerType() == PulsarMarkers.MarkerType.TXN_ABORT_VALUE) {
if (msgMetadata.hasMarkerType()
&& (msgMetadata.getMarkerType() == MarkerType.TXN_COMMIT_VALUE
|| msgMetadata.getMarkerType() == MarkerType.TXN_ABORT_VALUE)) {
MemoryRecords memoryRecords = MemoryRecords.withEndTransactionMarker(
baseOffset,
msgMetadata.getPublishTime(),
0,
msgMetadata.getTxnidMostBits(),
(short) msgMetadata.getTxnidLeastBits(),
new EndTransactionMarker(
msgMetadata.getMarkerType() == PulsarMarkers.MarkerType.TXN_COMMIT_VALUE
msgMetadata.getMarkerType() == MarkerType.TXN_COMMIT_VALUE
? ControlRecordType.COMMIT : ControlRecordType.ABORT, 0));
byteBuffer.put(memoryRecords.buffer());
return;
Expand Down Expand Up @@ -204,15 +188,13 @@ public MemoryRecords decode(final List<Entry> entries, final byte magic) {
log.debug(" processing message num - {} in batch", i);
}
try {
SingleMessageMetadata.Builder singleMessageMetadataBuilder = SingleMessageMetadata
.newBuilder();
final SingleMessageMetadata singleMessageMetadata = new SingleMessageMetadata();
ByteBuf singleMessagePayload = Commands.deSerializeSingleMessageInBatch(payload,
singleMessageMetadataBuilder, i, numMessages);
singleMessageMetadata, i, numMessages);

SingleMessageMetadata singleMessageMetadata = singleMessageMetadataBuilder.build();
Header[] headers = getHeadersFromMetadata(singleMessageMetadata.getPropertiesList());

final ByteBuffer value = (singleMessageMetadata.getNullValue())
final ByteBuffer value = (singleMessageMetadata.isNullValue())
? null
: ByteBufUtils.getNioBuffer(singleMessagePayload);
builder.appendWithOffset(
Expand All @@ -223,7 +205,6 @@ public MemoryRecords decode(final List<Entry> entries, final byte magic) {
value,
headers);
singleMessagePayload.release();
singleMessageMetadataBuilder.recycle();
} catch (IOException e) {
log.error("Meet IOException: {}", e);
throw new UncheckedIOException(e);
Expand All @@ -240,7 +221,6 @@ public MemoryRecords decode(final List<Entry> entries, final byte magic) {
headers);
}

msgMetadata.recycle();
payload.release();
entry.release();

Expand Down Expand Up @@ -276,9 +256,13 @@ private static MessageImpl<byte[]> recordToEntry(Record record) {
builder.value(null);
}

// sequence
// Following fields are required in `Commands.initBatchMessageMetadata`, but since we write to
// bookie directly, broker won't make use of them. So here we just set trivial values.
builder.getMetadataBuilder().setProducerName("");
if (record.sequence() >= 0) {
builder.sequenceId(record.sequence());
} else {
builder.sequenceId(0L);
}

// timestamp
Expand Down
26 changes: 16 additions & 10 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ByteBufUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,31 +18,37 @@
import io.netty.buffer.ByteBuf;
import java.nio.ByteBuffer;
import java.util.Base64;
import org.apache.pulsar.common.api.proto.PulsarApi;

import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.api.proto.SingleMessageMetadata;


/**
* Utils for ByteBuf operations.
*/
public class ByteBufUtils {

public static ByteBuffer getKeyByteBuffer(PulsarApi.SingleMessageMetadata messageMetadata) {
public static ByteBuffer getKeyByteBuffer(SingleMessageMetadata messageMetadata) {
if (messageMetadata.hasOrderingKey()) {
return messageMetadata.getOrderingKey().asReadOnlyByteBuffer();
return ByteBuffer.wrap(messageMetadata.getOrderingKey()).asReadOnlyBuffer();
}

String key = messageMetadata.getPartitionKey();
if (messageMetadata.hasPartitionKeyB64Encoded()) {
return ByteBuffer.wrap(Base64.getDecoder().decode(key));
if (messageMetadata.hasPartitionKey()) {
final String key = messageMetadata.getPartitionKey();
if (messageMetadata.hasPartitionKeyB64Encoded()) {
return ByteBuffer.wrap(Base64.getDecoder().decode(key)).asReadOnlyBuffer();
} else {
// for Base64 not encoded string, convert to UTF_8 chars
return ByteBuffer.wrap(key.getBytes(UTF_8));
}
} else {
// for Base64 not encoded string, convert to UTF_8 chars
return ByteBuffer.wrap(key.getBytes(UTF_8));
return ByteBuffer.allocate(0);
}
}

public static ByteBuffer getKeyByteBuffer(PulsarApi.MessageMetadata messageMetadata) {
public static ByteBuffer getKeyByteBuffer(MessageMetadata messageMetadata) {
if (messageMetadata.hasOrderingKey()) {
return messageMetadata.getOrderingKey().asReadOnlyByteBuffer();
return ByteBuffer.wrap(messageMetadata.getOrderingKey()).asReadOnlyBuffer();
}

String key = messageMetadata.getPartitionKey();
Expand Down
17 changes: 14 additions & 3 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MessageIdUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,19 @@
*/
package io.streamnative.pulsar.handlers.kop.utils;

import io.netty.buffer.ByteBuf;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.intercept.ManagedLedgerInterceptorImpl;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.Commands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -45,6 +48,14 @@ public static long getLogEndOffset(ManagedLedger managedLedger) {
return getCurrentOffset(managedLedger) + 1;
}

public static long getPublishTime(final ByteBuf byteBuf) {
final int readerIndex = byteBuf.readerIndex();
Commands.skipBrokerEntryMetadataIfExist(byteBuf);
final MessageMetadata metadata = Commands.parseMessageMetadata(byteBuf);
byteBuf.readerIndex(readerIndex);
return metadata.getPublishTime();
}

public static CompletableFuture<Long> getOffsetOfPosition(ManagedLedgerImpl managedLedger,
PositionImpl position,
boolean needCheckMore,
Expand All @@ -61,8 +72,8 @@ public void readEntryComplete(Entry entry, Object ctx) {
try {
if (needCheckMore) {
long offset = peekOffsetFromEntry(entry);
MessageImpl msg = MessageImpl.deserialize(entry.getDataBuffer());
if (msg.getPublishTime() >= timestamp) {
final long publishTime = getPublishTime(entry.getDataBuffer());
if (publishTime >= timestamp) {
future.complete(offset);
} else {
future.complete(offset + 1);
Expand Down
12 changes: 5 additions & 7 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/OffsetFinder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import com.google.common.base.Predicate;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
Expand All @@ -29,7 +30,6 @@
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.client.impl.MessageImpl;

/**
* given a timestamp find the first message (position) (published) at or before the timestamp.
Expand Down Expand Up @@ -59,17 +59,15 @@ public void findMessages(final long timestamp, AsyncCallbacks.FindEntryCallback
}

asyncFindNewestMatching(ManagedCursor.FindPositionConstraint.SearchAllAvailableEntries, entry -> {
MessageImpl msg = null;
if (entry == null) {
return false;
}
try {
msg = MessageImpl.deserialize(entry.getDataBuffer());
return msg.getPublishTime() <= timestamp;
return MessageIdUtils.getPublishTime(entry.getDataBuffer()) <= timestamp;
} catch (Exception e) {
log.error("[{}] Error deserialize message for message position find", managedLedger.getName(), e);
} finally {
entry.release();
if (msg != null) {
msg.recycle();
}
}
return false;
}, this, callback);
Expand Down
Loading