diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
index 9477af4f7846c..232f7dcaadcf4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
@@ -37,6 +37,7 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.nio.ByteBuffer;
 import java.nio.file.Files;
 import java.util.Arrays;
 import java.util.Collections;
@@ -74,19 +75,20 @@
 import org.apache.pulsar.common.io.SourceConfig;
 import org.apache.pulsar.common.nar.NarClassLoader;
 import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.policies.data.ClusterDataImpl;
 import org.apache.pulsar.common.policies.data.ConsumerStats;
 import org.apache.pulsar.common.policies.data.PublisherStats;
 import org.apache.pulsar.common.policies.data.SubscriptionStats;
 import org.apache.pulsar.common.policies.data.TenantInfo;
-import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.policies.data.TopicStats;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.functions.LocalRunner;
+import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory;
 import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig;
 import org.apache.pulsar.functions.utils.FunctionCommon;
+import org.apache.pulsar.io.core.Sink;
+import org.apache.pulsar.io.core.SinkContext;
 import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -851,7 +853,7 @@ public void testPulsarSourceLocalRunMultipleInstances() throws Throwable {
         runWithNarClassLoader(() -> testPulsarSourceLocalRun(null, 2));
     }
 
-    private void testPulsarSinkLocalRun(String jarFilePathUrl, int parallelism) throws Exception {
+    private void testPulsarSinkLocalRun(String jarFilePathUrl, int parallelism, String className) throws Exception {
         final String namespacePortion = "io";
         final String replNamespace = tenant + "/" + namespacePortion;
         final String sourceTopic = "persistent://" + replNamespace + "/input";
@@ -869,7 +871,9 @@ private void testPulsarSinkLocalRun(String jarFilePathUrl, int parallelism) thro
         SinkConfig sinkConfig = createSinkConfig(tenant, namespacePortion, sinkName, sourceTopic, subscriptionName);
 
         sinkConfig.setInputSpecs(Collections.singletonMap(sourceTopic, ConsumerConfig.builder().receiverQueueSize(1000).build()));
-        if (jarFilePathUrl == null || !jarFilePathUrl.endsWith(".nar")) {
+        if (className != null) {
+            sinkConfig.setClassName(className);
+        } else if (jarFilePathUrl == null || !jarFilePathUrl.endsWith(".nar")) {
             sinkConfig.setClassName("org.apache.pulsar.io.datagenerator.DataGeneratorPrintSink");
         }
 
@@ -935,6 +939,14 @@ private void testPulsarSinkLocalRun(String jarFilePathUrl, int parallelism) thro
                 if (m != null) {
                     metricsMap.put(m.tags.get("instance_id"), m);
                 }
+            } else if (line.startsWith("pulsar_sink_sink_exceptions_total")) {
+                Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(line);
+                assertFalse(metrics.isEmpty());
+                PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_sink_sink_exceptions_total");
+                if (m == null) {
+                    m = metrics.get("pulsar_sink_sink_exceptions_total_1min");
+                }
+                assertEquals(m.value, 0);
             }
         });
         Assert.assertEquals(metricsMap.size(), parallelism);
@@ -972,7 +984,11 @@ private void testPulsarSinkLocalRun(String jarFilePathUrl, int parallelism) thro
     }
 
     private void testPulsarSinkLocalRun(String jarFilePathUrl) throws Exception {
-        testPulsarSourceLocalRun(jarFilePathUrl, 1);
+        testPulsarSinkLocalRun(jarFilePathUrl, 1);
+    }
+
+    private void testPulsarSinkLocalRun(String jarFilePathUrl, int parallelism) throws Exception {
+        testPulsarSinkLocalRun(jarFilePathUrl, parallelism, null);
     }
 
     @Test(timeOut = 20000, groups = "builtin")
@@ -1001,6 +1017,31 @@ public void testPulsarSinkStatsMultipleInstances() throws Throwable {
         runWithNarClassLoader(() -> testPulsarSinkLocalRun(null, 2));
     }
 
+    public static class StatsNullSink implements Sink<ByteBuffer> {
+        volatile long bytesTotal = 0;
+
+        @Override
+        public void open(Map map, final SinkContext sinkContext) throws Exception {
+
+        }
+
+        @Override
+        public void write(Record<ByteBuffer> record) throws Exception {
+            bytesTotal += record.getValue().capacity();
+            record.ack();
+        }
+
+        @Override
+        public void close() throws Exception {
+
+        }
+    }
+
+    @Test
+    public void test() throws Throwable{
+        runWithNarClassLoader(() -> testPulsarSinkLocalRun(null, 1, StatsNullSink.class.getName()));
+    }
+
     private void runWithNarClassLoader(Assert.ThrowingRunnable throwingRunnable) throws Throwable {
         ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader();
         try (NarClassLoader classLoader = NarClassLoader.getFromArchive(getPulsarIODataGeneratorNar(), Collections.emptySet(), originalClassLoader, NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR)) {
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
index caf7ff8dbf803..dcd424e4e71d4 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
@@ -145,7 +145,11 @@ private static <T> Schema<T> newSchemaInstance(Class<T> clazz, SchemaType type)
     private static <T> Schema<T> newSchemaInstance(Class<T> clazz, SchemaType type, ConsumerConfig conf) {
         switch (type) {
         case NONE:
-            return (Schema<T>) Schema.BYTES;
+            if (ByteBuffer.class.isAssignableFrom(clazz)) {
+                return (Schema<T>) Schema.BYTEBUFFER;
+            } else {
+                return (Schema<T>) Schema.BYTES;
+            }
 
         case AUTO_CONSUME:
         case AUTO: