Bugfix - release and recycle on discarded messages
Don't leak resources when a message is being discarded.
*Modifications*
- Fix missing release() and recycle() for discarded message on
receiveIndividualMessagesFromBatch method.
- Fix argument missing of debug logging {}-placeholder.
- Fix unnecessary variable reference `payload` on messageReceived().
@@ -747,17 +747,15 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
messageId.getEntryId());
}
- MessageMetadata msgMetadata = null;
- ByteBuf payload = headersAndPayload;
-
if (!verifyChecksum(headersAndPayload, messageId)) {
// discard message with checksum error
discardCorruptedMessage(messageId, cnx, ValidationError.ChecksumMismatch);
return;
}
+ MessageMetadata msgMetadata;
try {
- msgMetadata = Commands.parseMessageMetadata(payload);
+ msgMetadata = Commands.parseMessageMetadata(headersAndPayload);
} catch (Throwable t) {
discardCorruptedMessage(messageId, cnx, ValidationError.ChecksumMismatch);
return;
@@ -769,14 +767,14 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (acknowledgmentsGroupingTracker.isDuplicate(msgId)) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Ignoring message as it was already being acked earlier by same consumer {}/{}",
- topic, subscription, msgId);
+ topic, subscription, consumerName, msgId);
}
increaseAvailablePermits(cnx, numMessages);
return;
}
- ByteBuf decryptedPayload = decryptPayloadIfNeeded(messageId, msgMetadata, payload, cnx);
+ ByteBuf decryptedPayload = decryptPayloadIfNeeded(messageId, msgMetadata, headersAndPayload, cnx);
boolean isMessageUndecryptable = isMessageUndecryptable(msgMetadata);
@@ -950,6 +948,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
log.debug("[{}] [{}] Ignoring message from before the startMessageId", subscription,
consumerName);
}
+ singleMessagePayload.release();
+ singleMessageMetadataBuilder.recycle();
++skippedMessages;
continue;
