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 2021/10/05 05:29:24 UTC
[GitHub] [pulsar] BewareMyPower commented on a change in pull request #12088: [PIP 96] Add message payload processor for Pulsar client
BewareMyPower commented on a change in pull request #12088:
URL: https://github.com/apache/pulsar/pull/12088#discussion_r721899375
##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
##########
@@ -996,6 +996,130 @@ void activeConsumerChanged(boolean isActive) {
});
}
+ protected boolean isBatch(MessageMetadata messageMetadata) {
+ // if message is not decryptable then it can't be parsed as a batch-message. so, add EncyrptionCtx to message
+ // and return undecrypted payload
+ return !isMessageUndecryptable(messageMetadata) &&
+ (messageMetadata.hasNumMessagesInBatch() || messageMetadata.getNumMessagesInBatch() != 1);
+ }
+
+ protected <U> MessageImpl<U> newSingleMessage(final int index,
+ final int numMessages,
+ final BrokerEntryMetadata brokerEntryMetadata,
+ final MessageMetadata msgMetadata,
+ final SingleMessageMetadata singleMessageMetadata,
+ final ByteBuf payload,
+ final MessageIdImpl messageId,
+ final Schema<U> schema,
+ final boolean containMetadata,
+ final BitSetRecyclable ackBitSet,
+ final BatchMessageAcker acker,
+ final int redeliveryCount) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] [{}] processing message num - {} in batch", subscription, consumerName, index);
+ }
+
+ ByteBuf singleMessagePayload = null;
+ try {
+ if (containMetadata) {
+ singleMessagePayload =
+ Commands.deSerializeSingleMessageInBatch(payload, singleMessageMetadata, index, numMessages);
+ }
+
+ if (isSameEntry(messageId) && isPriorBatchIndex(index)) {
+ // If we are receiving a batch message, we need to discard messages that were prior
+ // to the startMessageId
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] [{}] Ignoring message from before the startMessageId: {}", subscription,
+ consumerName, startMessageId);
+ }
+ return null;
+ }
+
+ if (singleMessageMetadata != null && singleMessageMetadata.isCompactedOut()) {
+ // message has been compacted out, so don't send to the user
+ return null;
+ }
+
+ if (ackBitSet != null && !ackBitSet.get(index)) {
+ return null;
+ }
+
+ BatchMessageIdImpl batchMessageIdImpl = new BatchMessageIdImpl(messageId.getLedgerId(),
+ messageId.getEntryId(), getPartitionIndex(), index, numMessages, acker);
+
+ final ByteBuf payloadBuffer = (singleMessagePayload != null) ? singleMessagePayload : payload;
+ return MessageImpl.create(topicName.toString(), batchMessageIdImpl,
+ msgMetadata, singleMessageMetadata, payloadBuffer,
+ createEncryptionContext(msgMetadata), cnx(), schema, redeliveryCount, poolMessages
+ ).setBrokerEntryMetadata(brokerEntryMetadata);
+ } catch (IOException | IllegalStateException e) {
+ throw new IllegalStateException(e);
+ } finally {
+ if (singleMessagePayload != null) {
+ singleMessagePayload.release();
+ }
+ }
+ }
+
+ protected <U> MessageImpl<U> newMessage(final MessageIdImpl messageId,
+ final BrokerEntryMetadata brokerEntryMetadata,
+ final MessageMetadata messageMetadata,
+ final ByteBuf payload,
+ final Schema<U> schema,
+ final int redeliveryCount) {
+ return MessageImpl.create(topicName.toString(), messageId, messageMetadata, payload,
+ createEncryptionContext(messageMetadata), cnx(), schema, redeliveryCount, poolMessages
+ ).setBrokerEntryMetadata(brokerEntryMetadata);
+ }
+
+ private void executeNotifyCallback(final MessageImpl<T> message) {
+ // Enqueue the message so that it can be retrieved when application calls receive()
+ // if the conf.getReceiverQueueSize() is 0 then discard message if no one is waiting for it.
+ // if asyncReceive is waiting then notify callback without adding to incomingMessages queue
+ internalPinnedExecutor.execute(() -> {
+ if (hasNextPendingReceive()) {
+ notifyPendingReceivedCallback(message, null);
+ } else if (enqueueMessageAndCheckBatchReceive(message) && hasPendingBatchReceive()) {
+ notifyPendingBatchReceivedCallBack();
+ }
+ });
+ }
+
+ private void processPayloadByProcessor(final BrokerEntryMetadata brokerEntryMetadata,
+ final MessageMetadata messageMetadata,
+ final ByteBuf byteBuf,
+ final MessageIdImpl messageId,
+ final Schema<T> schema,
+ final int redeliveryCount,
+ final List<Long> ackSet) {
+ final MessagePayloadImpl payload = MessagePayloadImpl.create(byteBuf);
+ final MessagePayloadContextImpl entryContext = MessagePayloadContextImpl.get(
+ brokerEntryMetadata, messageMetadata, messageId, this, redeliveryCount, ackSet);
+ final AtomicInteger skippedMessages = new AtomicInteger(0);
+ try {
+ conf.getPayloadProcessor().process(payload, entryContext, schema, message -> {
+ if (message == null) {
+ skippedMessages.incrementAndGet();
+ }
+ executeNotifyCallback((MessageImpl<T>) message);
+ });
Review comment:
Nice point. I forgot to skip this call when I refactored the iterable interface to current implementation.
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org