KevinLiLu commented on code in PR #21798:
URL: https://github.com/apache/pulsar/pull/21798#discussion_r1435873119
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java:
##########
@@ -263,6 +263,29 @@ public boolean checkAndStartPublish(long producerId, long
sequenceId, ByteBuf he
}
}
+ if (topic.isPersistent()) {
+ PersistentTopic pTopic = (PersistentTopic) topic;
+ if (pTopic.isDelayedDeliveryEnabled()) {
+ long maxDeliveryDelayInMs =
pTopic.getDelayedDeliveryMaxDelayInMillis();
+ if (maxDeliveryDelayInMs > 0) {
+ headersAndPayload.markReaderIndex();
+ MessageMetadata msgMetadata =
Commands.parseMessageMetadata(headersAndPayload);
+ headersAndPayload.resetReaderIndex();
+ if (msgMetadata.hasDeliverAtTime()
+ && msgMetadata.getDeliverAtTime() -
msgMetadata.getPublishTime() > maxDeliveryDelayInMs) {
+ cnx.execute(() -> {
+ cnx.getCommandSender().sendSendError(producerId,
sequenceId,
+ ServerError.NotAllowedError,
+ String.format("Exceeds max allowed
delivery delay of %s milliseconds",
+ maxDeliveryDelayInMs));
+ cnx.completedSendOperation(false,
headersAndPayload.readableBytes());
+ });
+ return false;
+ }
Review Comment:
The other error checks in the `checkAndStartPublish` method (right above my
change) are calling `cnx.getCommandSender().sendSendError(...)` and then
`cnx.completedSendOperation(...)` immediately.
For instance, the encryption error check (when encryption is required but
the producer does not encrypt the message):
https://github.com/apache/pulsar/blob/32f3577a735581096d85aa961d7df45b9ae9b6f9/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java#L248-L264
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
