Fix `messageQueue` release message issue.
@@ -726,19 +726,17 @@ public class PulsarRecordCursor implements RecordCursor {
public void close() {
log.info("Closing cursor record");
- if (currentMessage != null) {
- currentMessage.release();
- }
-
- if (messageQueue != null) {
- messageQueue.drain(RawMessage::release);
- }
-
if (deserializeEntries != null) {
deserializeEntries.close().whenComplete((r, t) -> {
if (entryQueue != null) {
entryQueue.drain(Entry::release);
}
+ if (messageQueue != null) {
+ messageQueue.drain(RawMessage::release);
+ }
+ if (currentMessage != null) {
+ currentMessage.release();
+ }
});
}
