Skip to content

Support set publish time on broker side #3267

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

Closed
lovelle wants to merge 1 commit into from
Closed

Support set publish time on broker side #3267

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
14 changes: 12 additions & 2 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.api.proto.PulsarApi.ServerError;
import org.apache.pulsar.common.naming.Constants;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.NonPersistentPublisherStats;
import org.apache.pulsar.common.policies.data.PublisherStats;
Expand Down Expand Up @@ -148,12 +149,17 @@ public void publishMessage(long producerId, long sequenceId, ByteBuf headersAndP
return;
}

if (topic.isEncryptionRequired()) {
headersAndPayload.markReaderIndex();
MessageMetadata msgMetadata = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.resetReaderIndex();

if (msgMetadata.getPublishTime() == Constants.PUBLISH_TIME_UNSET_MS) {

@merlimat merlimat Dec 30, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of constants, we should just use the hasPublishTime() method that will report if the field was set in the serizalized protobuf binary.

@lovelle lovelle Dec 30, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, I didn't explained why I did it that way.

The reason I used a constant of 13 digits size is because the method modifyMessageMetadata() should be able to just update this value without realloc the entire buffer due to being with different size, the problem with hasPublishTime() is that if it is false then publish time on metadata is literally '0' and the new epoch will be of 13 digits length.

The alternative approach I can think to get rid of the constant definition is to set publish time field on protobuf with the default value of 13 digits and testing if publish time has such default comparing with a new metadata created.
Maybe proto3 solved this? not really sure.

headersAndPayload.markReaderIndex();
MessageMetadata msgMetadata = Commands.parseMessageMetadata(headersAndPayload);
Commands.modifyMessageMetadata(headersAndPayload, newPublishOnMetadata(msgMetadata));
headersAndPayload.resetReaderIndex();
}

if (topic.isEncryptionRequired()) {
// Check whether the message is encrypted or not
if (msgMetadata.getEncryptionKeysCount() < 1) {
log.warn("[{}] Messages must be encrypted", getTopic().getName());
Expand All @@ -171,6 +177,10 @@ public void publishMessage(long producerId, long sequenceId, ByteBuf headersAndP
MessagePublishContext.get(this, sequenceId, msgIn, headersAndPayload.readableBytes(), batchSize));
}

private static MessageMetadata newPublishOnMetadata(MessageMetadata msgMetadata) {
return msgMetadata.toBuilder().setPublishTime(System.currentTimeMillis()).build();
}

private boolean verifyChecksum(ByteBuf headersAndPayload) {
if (hasChecksum(headersAndPayload)) {
int readerIndex = headersAndPayload.readerIndex();
Expand Down
3 changes: 2 additions & 1 deletion pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
import org.apache.pulsar.common.api.proto.PulsarApi.ProtocolVersion;
import org.apache.pulsar.common.compression.CompressionCodec;
import org.apache.pulsar.common.compression.CompressionCodecProvider;
import org.apache.pulsar.common.naming.Constants;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.common.util.DateFormatter;
Expand Down Expand Up @@ -344,7 +345,7 @@ public void sendAsync(Message<T> message, SendCallback callback) {
sequenceId = msgMetadataBuilder.getSequenceId();
}
if (!msgMetadataBuilder.hasPublishTime()) {
msgMetadataBuilder.setPublishTime(System.currentTimeMillis());
msgMetadataBuilder.setPublishTime(Constants.PUBLISH_TIME_UNSET_MS);

checkArgument(!msgMetadataBuilder.hasProducerName());

Expand Down
34 changes: 34 additions & 0 deletions pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,40 @@ public static void skipChecksumIfPresent(ByteBuf buffer) {
}
}

public static void modifyMessageMetadata(ByteBuf buffer, MessageMetadata newMsgMetadata) {
try {
int checksumReaderIndex = buffer.readerIndex();
skipChecksumIfPresent(buffer);
int metadataSize = (int) buffer.readUnsignedInt();
int newMetadataSize = newMsgMetadata.getSerializedSize();

if (metadataSize != newMetadataSize) {
throw new RuntimeException("Write new metadata with different size from previous is not yet supported");
}

buffer.markWriterIndex();
buffer.writerIndex(buffer.readerIndex());
ByteBufCodedOutputStream outputStream = ByteBufCodedOutputStream.get(buffer);
newMsgMetadata.writeTo(outputStream);
buffer.resetWriterIndex();

buffer.readerIndex(checksumReaderIndex);
if (hasChecksum(buffer)) {
int currentChecksum = readChecksum(buffer);
int computedChecksum = computeChecksum(buffer);
// set new computed checksum
if (currentChecksum != computedChecksum) {
buffer.setInt(checksumReaderIndex+2, computedChecksum);
}
}

outputStream.recycle();
newMsgMetadata.recycle();
} catch (IOException e) {
throw new RuntimeException(e);
}
}

public static MessageMetadata parseMessageMetadata(ByteBuf buffer) {
try {
// initially reader-index may point to start_of_checksum : increment reader-index to start_of_metadata to parse
Expand Down
1 change: 1 addition & 0 deletions pulsar-common/src/main/java/org/apache/pulsar/common/naming/Constants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

public class Constants {

public static final long PUBLISH_TIME_UNSET_MS = 1000000000000L;
public static final String GLOBAL_CLUSTER = "global";

private Constants() {}
Expand Down