Move message release logic to correct space.

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/emitter/PulsarRecordEmitter.java

@@ -53,6 +53,9 @@ public void emitRecord(
 
         deserializationSchema.deserialize(element, sourceOutputWrapper);
         splitState.setLatestConsumedId(element.getMessageId());
+
+        // Recycle message.
+        element.release();
     }
 
     private static class SourceOutputWrapper<T> implements Collector<T> {

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java

@@ -71,9 +71,6 @@ protected Message<byte[]> pollMessage(Duration timeout) throws PulsarClientExcep
     protected void finishedPollMessage(Message<?> message) {
         // Nothing to do here.
         LOG.debug("Finished polling message {}", message);
-
-        // Release message
-        message.release();
     }
 
     @Override

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java

@@ -111,9 +111,6 @@ protected void finishedPollMessage(Message<?> message) {
         if (sourceConfiguration.isEnableAutoAcknowledgeMessage()) {
             sneakyClient(() -> pulsarConsumer.acknowledge(message));
         }
-
-        // Release message
-        message.release();
     }
 
     @Override