diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java
index 2a857c4f25451..e5299cb12b5a6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java
@@ -23,6 +23,7 @@
 import static org.apache.commons.lang.StringUtils.defaultIfEmpty;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Charsets;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.time.Clock;
 import java.util.List;
@@ -143,8 +144,14 @@ public void postSchema(PostSchemaPayload payload, boolean authoritative, AsyncRe
             }
             byte[] data;
             if (SchemaType.KEY_VALUE.name().equals(payload.getType())) {
-                data = DefaultImplementation
-                        .convertKeyValueDataStringToSchemaInfoSchema(payload.getSchema().getBytes(Charsets.UTF_8));
+                try {
+                    data = DefaultImplementation.getDefaultImplementation()
+                            .convertKeyValueDataStringToSchemaInfoSchema(payload.getSchema().getBytes(Charsets.UTF_8));
+                } catch (IOException conversionError) {
+                    log.error("[{}] Failed to post schema for topic {}", clientAppId(), topicName, conversionError);
+                    response.resume(Response.serverError().build());
+                    return;
+                }
             } else {
                 data = payload.getSchema().getBytes(Charsets.UTF_8);
             }
@@ -241,16 +248,21 @@ protected String domain() {
     }
 
     private static GetSchemaResponse convertSchemaAndMetadataToGetSchemaResponse(SchemaAndMetadata schemaAndMetadata) {
-        String schemaData;
-        if (schemaAndMetadata.schema.getType() == SchemaType.KEY_VALUE) {
-            schemaData = DefaultImplementation.convertKeyValueSchemaInfoDataToString(
-                    DefaultImplementation.decodeKeyValueSchemaInfo(schemaAndMetadata.schema.toSchemaInfo()));
-        } else {
-            schemaData = new String(schemaAndMetadata.schema.getData(), UTF_8);
+        try {
+            String schemaData;
+            if (schemaAndMetadata.schema.getType() == SchemaType.KEY_VALUE) {
+                schemaData = DefaultImplementation.getDefaultImplementation().convertKeyValueSchemaInfoDataToString(
+                        DefaultImplementation.getDefaultImplementation()
+                                .decodeKeyValueSchemaInfo(schemaAndMetadata.schema.toSchemaInfo()));
+            } else {
+                schemaData = new String(schemaAndMetadata.schema.getData(), UTF_8);
+            }
+            return GetSchemaResponse.builder().version(getLongSchemaVersion(schemaAndMetadata.version))
+                    .type(schemaAndMetadata.schema.getType()).timestamp(schemaAndMetadata.schema.getTimestamp())
+                    .data(schemaData).properties(schemaAndMetadata.schema.getProps()).build();
+        } catch (IOException conversionError) {
+            throw new RuntimeException(conversionError);
         }
-        return GetSchemaResponse.builder().version(getLongSchemaVersion(schemaAndMetadata.version))
-                .type(schemaAndMetadata.schema.getType()).timestamp(schemaAndMetadata.schema.getTimestamp())
-                .data(schemaData).properties(schemaAndMetadata.schema.getProps()).build();
     }
 
     private static void handleGetSchemaResponse(AsyncResponse response, SchemaAndMetadata schema, Throwable error) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageIdSerializationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageIdSerializationTest.java
index 01661ce84eea2..295c7803372f9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageIdSerializationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageIdSerializationTest.java
@@ -19,7 +19,7 @@
 package org.apache.pulsar.broker.service;
 
 import static org.testng.Assert.assertEquals;
-
+import java.io.IOException;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.testng.annotations.Test;
@@ -46,7 +46,7 @@ public void testProtobufSerializationNull() throws Exception {
         MessageId.fromByteArray(null);
     }
 
-    @Test(expectedExceptions = RuntimeException.class)
+    @Test(expectedExceptions = IOException.class)
     void testProtobufSerializationEmpty() throws Exception {
         MessageId.fromByteArray(new byte[0]);
     }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
index f971b5a770cca..d7cb6c9cc3220 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
@@ -66,7 +66,6 @@
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.policies.data.ClusterDataImpl;
 import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
@@ -548,7 +547,7 @@ private void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, Subscri
             }
 
             try {
-                consumer.acknowledgeCumulativeAsync(DefaultImplementation
+                consumer.acknowledgeCumulativeAsync(DefaultImplementation.getDefaultImplementation()
                         .newMessageId(((MessageIdImpl) message.getMessageId()).getLedgerId(),
                                 ((MessageIdImpl) message.getMessageId()).getEntryId() - 1, -1),
                         abortTxn).get();
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java
index 4408ae21b50f0..a072acd8b73a4 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SchemasImpl.java
@@ -19,6 +19,7 @@
 package org.apache.pulsar.client.admin.internal;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -438,8 +439,12 @@ static SchemaInfo convertGetSchemaResponseToSchemaInfo(TopicName tn,
 
         byte[] schema;
         if (response.getType() == SchemaType.KEY_VALUE) {
-            schema = DefaultImplementation.convertKeyValueDataStringToSchemaInfoSchema(
-                    response.getData().getBytes(UTF_8));
+            try {
+                schema = DefaultImplementation.getDefaultImplementation().convertKeyValueDataStringToSchemaInfoSchema(
+                        response.getData().getBytes(UTF_8));
+            } catch (IOException conversionError) {
+                throw new RuntimeException(conversionError);
+            }
         } else {
             schema = response.getData().getBytes(UTF_8);
         }
@@ -466,28 +471,31 @@ static SchemaInfoWithVersion convertGetSchemaResponseToSchemaInfoWithVersion(Top
 
 
     // the util function exists for backward compatibility concern
-    static String convertSchemaDataToStringLegacy(SchemaInfo schemaInfo) {
+    static String convertSchemaDataToStringLegacy(SchemaInfo schemaInfo) throws IOException {
         byte[] schemaData = schemaInfo.getSchema();
         if (null == schemaInfo.getSchema()) {
             return "";
         }
 
         if (schemaInfo.getType() == SchemaType.KEY_VALUE) {
-           return DefaultImplementation.convertKeyValueSchemaInfoDataToString(
-                   DefaultImplementation.decodeKeyValueSchemaInfo(schemaInfo));
+           return DefaultImplementation.getDefaultImplementation().convertKeyValueSchemaInfoDataToString(
+                   DefaultImplementation.getDefaultImplementation().decodeKeyValueSchemaInfo(schemaInfo));
         }
 
         return new String(schemaData, UTF_8);
     }
 
     static PostSchemaPayload convertSchemaInfoToPostSchemaPayload(SchemaInfo schemaInfo) {
-
-        PostSchemaPayload payload = new PostSchemaPayload();
-        payload.setType(schemaInfo.getType().name());
-        payload.setProperties(schemaInfo.getProperties());
-        // for backward compatibility concern, we convert `bytes` to `string`
-        // we can consider fixing it in a new version of rest endpoint
-        payload.setSchema(convertSchemaDataToStringLegacy(schemaInfo));
-        return payload;
+        try {
+            PostSchemaPayload payload = new PostSchemaPayload();
+            payload.setType(schemaInfo.getType().name());
+            payload.setProperties(schemaInfo.getProperties());
+            // for backward compatibility concern, we convert `bytes` to `string`
+            // we can consider fixing it in a new version of rest endpoint
+            payload.setSchema(convertSchemaDataToStringLegacy(schemaInfo));
+            return payload;
+        } catch (IOException conversionError) {
+            throw new RuntimeException(conversionError);
+        }
     }
 }
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationFactory.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationFactory.java
index a244c3d2c8e4c..4de12b2eff2f0 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationFactory.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationFactory.java
@@ -41,7 +41,7 @@ public final class AuthenticationFactory {
      * @return the Authentication object initialized with the token credentials
      */
     public static Authentication token(String token) {
-        return DefaultImplementation.newAuthenticationToken(token);
+        return DefaultImplementation.getDefaultImplementation().newAuthenticationToken(token);
     }
 
     /**
@@ -52,7 +52,7 @@ public static Authentication token(String token) {
      * @return the Authentication object initialized with the token credentials
      */
     public static Authentication token(Supplier<String> tokenSupplier) {
-        return DefaultImplementation.newAuthenticationToken(tokenSupplier);
+        return DefaultImplementation.getDefaultImplementation().newAuthenticationToken(tokenSupplier);
     }
 
     // CHECKSTYLE.OFF: MethodName
@@ -67,7 +67,7 @@ public static Authentication token(Supplier<String> tokenSupplier) {
      * @return the Authentication object initialized with the TLS credentials
      */
     public static Authentication TLS(String certFilePath, String keyFilePath) {
-        return DefaultImplementation.newAuthenticationTLS(certFilePath, keyFilePath);
+        return DefaultImplementation.getDefaultImplementation().newAuthenticationTLS(certFilePath, keyFilePath);
     }
 
     // CHECKSTYLE.ON: MethodName
@@ -86,7 +86,7 @@ public static Authentication TLS(String certFilePath, String keyFilePath) {
     public static Authentication create(String authPluginClassName, String authParamsString)
             throws UnsupportedAuthenticationException {
         try {
-            return DefaultImplementation.createAuthentication(authPluginClassName, authParamsString);
+            return DefaultImplementation.getDefaultImplementation().createAuthentication(authPluginClassName, authParamsString);
         } catch (Throwable t) {
             throw new UnsupportedAuthenticationException(t);
         }
@@ -103,7 +103,7 @@ public static Authentication create(String authPluginClassName, String authParam
     public static Authentication create(String authPluginClassName, Map<String, String> authParams)
             throws UnsupportedAuthenticationException {
         try {
-            return DefaultImplementation.createAuthentication(authPluginClassName, authParams);
+            return DefaultImplementation.getDefaultImplementation().createAuthentication(authPluginClassName, authParams);
         } catch (Throwable t) {
             throw new UnsupportedAuthenticationException(t);
         }
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/BatcherBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/BatcherBuilder.java
index d659a952e8fcb..34d8375fbcead 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/BatcherBuilder.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/BatcherBuilder.java
@@ -19,6 +19,7 @@
 package org.apache.pulsar.client.api;
 
 import java.io.Serializable;
+
 import org.apache.pulsar.client.internal.DefaultImplementation;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
@@ -39,7 +40,7 @@ public interface BatcherBuilder extends Serializable {
      * <p>batched into single batch message:
      * [(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]
      */
-    BatcherBuilder DEFAULT = DefaultImplementation.newDefaultBatcherBuilder();
+    BatcherBuilder DEFAULT = DefaultImplementation.getDefaultImplementation().newDefaultBatcherBuilder();
 
     /**
      * Key based batch message container.
@@ -50,7 +51,7 @@ public interface BatcherBuilder extends Serializable {
      * <p>batched into multiple batch messages:
      * [(k1, v1), (k1, v2), (k1, v3)], [(k2, v1), (k2, v2), (k2, v3)], [(k3, v1), (k3, v2), (k3, v3)]
      */
-    BatcherBuilder KEY_BASED = DefaultImplementation.newKeyBasedBatcherBuilder();
+    BatcherBuilder KEY_BASED = DefaultImplementation.getDefaultImplementation().newKeyBasedBatcherBuilder();
 
     /**
      * Build a new batch message container.
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageId.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageId.java
index 7b275c448d3df..bf8defefe0c61 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageId.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageId.java
@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 import java.io.Serializable;
+
 import org.apache.pulsar.client.internal.DefaultImplementation;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
@@ -54,7 +55,7 @@ public interface MessageId extends Comparable<MessageId>, Serializable {
      * @throws IOException if the de-serialization fails
      */
     static MessageId fromByteArray(byte[] data) throws IOException {
-        return DefaultImplementation.newMessageIdFromByteArray(data);
+        return DefaultImplementation.getDefaultImplementation().newMessageIdFromByteArray(data);
     }
 
     /**
@@ -70,7 +71,7 @@ static MessageId fromByteArray(byte[] data) throws IOException {
      * @throws IOException if the de-serialization fails
      */
     static MessageId fromByteArrayWithTopic(byte[] data, String topicName) throws IOException {
-        return DefaultImplementation.newMessageIdFromByteArrayWithTopic(data, topicName);
+        return DefaultImplementation.getDefaultImplementation().newMessageIdFromByteArrayWithTopic(data, topicName);
     }
 
     // CHECKSTYLE.OFF: ConstantName
@@ -78,12 +79,12 @@ static MessageId fromByteArrayWithTopic(byte[] data, String topicName) throws IO
     /**
      * MessageId that represents the oldest message available in the topic.
      */
-    MessageId earliest = DefaultImplementation.newMessageId(-1, -1, -1);
+    MessageId earliest = DefaultImplementation.getDefaultImplementation().newMessageId(-1, -1, -1);
 
     /**
      * MessageId that represents the next message published in the topic.
      */
-    MessageId latest = DefaultImplementation.newMessageId(Long.MAX_VALUE, Long.MAX_VALUE, -1);
+    MessageId latest = DefaultImplementation.getDefaultImplementation().newMessageId(Long.MAX_VALUE, Long.MAX_VALUE, -1);
 
     // CHECKSTYLE.ON: ConstantName
 }
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
index 5cf53090455b4..5f1a18b0e0235 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
@@ -52,7 +52,7 @@ public interface PulsarClient extends Closeable {
      * @since 2.0.0
      */
     static ClientBuilder builder() {
-        return DefaultImplementation.newClientBuilder();
+        return DefaultImplementation.getDefaultImplementation().newClientBuilder();
     }
 
     /**
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
index 6cf9ccdc794be..6803187521adf 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
@@ -18,7 +18,7 @@
  */
 package org.apache.pulsar.client.api;
 
-import static org.apache.pulsar.client.internal.DefaultImplementation.getBytes;
+import static org.apache.pulsar.client.internal.PulsarClientImplementationBinding.getBytes;
 import java.nio.ByteBuffer;
 import java.sql.Time;
 import java.sql.Timestamp;
@@ -174,7 +174,7 @@ default void configureSchemaInfo(String topic, String componentName,
     /**
      * Schema that doesn't perform any encoding on the message payloads. Accepts a byte array and it passes it through.
      */
-    Schema<byte[]> BYTES = DefaultImplementation.newBytesSchema();
+    Schema<byte[]> BYTES = DefaultImplementation.getDefaultImplementation().newBytesSchema();
 
     /**
      * Return the native schema that is wrapped by Pulsar API.
@@ -188,79 +188,79 @@ default Optional<Object> getNativeSchema() {
     /**
      * ByteBuffer Schema.
      */
-    Schema<ByteBuffer> BYTEBUFFER = DefaultImplementation.newByteBufferSchema();
+    Schema<ByteBuffer> BYTEBUFFER = DefaultImplementation.getDefaultImplementation().newByteBufferSchema();
 
     /**
      * Schema that can be used to encode/decode messages whose values are String. The payload is encoded with UTF-8.
      */
-    Schema<String> STRING = DefaultImplementation.newStringSchema();
+    Schema<String> STRING = DefaultImplementation.getDefaultImplementation().newStringSchema();
 
     /**
      * INT8 Schema.
      */
-    Schema<Byte> INT8 = DefaultImplementation.newByteSchema();
+    Schema<Byte> INT8 = DefaultImplementation.getDefaultImplementation().newByteSchema();
 
     /**
      * INT16 Schema.
      */
-    Schema<Short> INT16 = DefaultImplementation.newShortSchema();
+    Schema<Short> INT16 = DefaultImplementation.getDefaultImplementation().newShortSchema();
 
     /**
      * INT32 Schema.
      */
-    Schema<Integer> INT32 = DefaultImplementation.newIntSchema();
+    Schema<Integer> INT32 = DefaultImplementation.getDefaultImplementation().newIntSchema();
 
     /**
      * INT64 Schema.
      */
-    Schema<Long> INT64 = DefaultImplementation.newLongSchema();
+    Schema<Long> INT64 = DefaultImplementation.getDefaultImplementation().newLongSchema();
 
     /**
      * Boolean Schema.
      */
-    Schema<Boolean> BOOL = DefaultImplementation.newBooleanSchema();
+    Schema<Boolean> BOOL = DefaultImplementation.getDefaultImplementation().newBooleanSchema();
 
     /**
      * Float Schema.
      */
-    Schema<Float> FLOAT = DefaultImplementation.newFloatSchema();
+    Schema<Float> FLOAT = DefaultImplementation.getDefaultImplementation().newFloatSchema();
 
     /**
      * Double Schema.
      */
-    Schema<Double> DOUBLE = DefaultImplementation.newDoubleSchema();
+    Schema<Double> DOUBLE = DefaultImplementation.getDefaultImplementation().newDoubleSchema();
 
     /**
      * Date Schema.
      */
-    Schema<Date> DATE = DefaultImplementation.newDateSchema();
+    Schema<Date> DATE = DefaultImplementation.getDefaultImplementation().newDateSchema();
 
     /**
      * Time Schema.
      */
-    Schema<Time> TIME = DefaultImplementation.newTimeSchema();
+    Schema<Time> TIME = DefaultImplementation.getDefaultImplementation().newTimeSchema();
 
     /**
      * Timestamp Schema.
      */
-    Schema<Timestamp> TIMESTAMP = DefaultImplementation.newTimestampSchema();
+    Schema<Timestamp> TIMESTAMP = DefaultImplementation.getDefaultImplementation().newTimestampSchema();
 
     /**
      * Instant Schema.
      */
-    Schema<Instant> INSTANT = DefaultImplementation.newInstantSchema();
+    Schema<Instant> INSTANT = DefaultImplementation.getDefaultImplementation().newInstantSchema();
     /**
      * LocalDate Schema.
      */
-    Schema<LocalDate> LOCAL_DATE = DefaultImplementation.newLocalDateSchema();
+    Schema<LocalDate> LOCAL_DATE = DefaultImplementation.getDefaultImplementation().newLocalDateSchema();
     /**
      * LocalTime Schema.
      */
-    Schema<LocalTime> LOCAL_TIME = DefaultImplementation.newLocalTimeSchema();
+    Schema<LocalTime> LOCAL_TIME = DefaultImplementation.getDefaultImplementation().newLocalTimeSchema();
     /**
      * LocalDateTime Schema.
      */
-    Schema<LocalDateTime> LOCAL_DATE_TIME = DefaultImplementation.newLocalDateTimeSchema();
+    Schema<LocalDateTime> LOCAL_DATE_TIME = DefaultImplementation.getDefaultImplementation().newLocalDateTimeSchema();
 
     // CHECKSTYLE.OFF: MethodName
 
@@ -271,7 +271,7 @@ default Optional<Object> getNativeSchema() {
      * @return a Schema instance
      */
     static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF(Class<T> clazz) {
-        return DefaultImplementation.newProtobufSchema(SchemaDefinition.builder().withPojo(clazz).build());
+        return DefaultImplementation.getDefaultImplementation().newProtobufSchema(SchemaDefinition.builder().withPojo(clazz).build());
     }
 
     /**
@@ -281,7 +281,7 @@ static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF(Cla
      * @return a Schema instance
      */
     static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF(SchemaDefinition<T> schemaDefinition) {
-        return DefaultImplementation.newProtobufSchema(schemaDefinition);
+        return DefaultImplementation.getDefaultImplementation().newProtobufSchema(schemaDefinition);
     }
 
     /**
@@ -291,7 +291,7 @@ static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF(Sch
      * @return a Schema instance
      */
     static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF_NATIVE(Class<T> clazz) {
-        return DefaultImplementation.newProtobufNativeSchema(SchemaDefinition.builder().withPojo(clazz).build());
+        return DefaultImplementation.getDefaultImplementation().newProtobufNativeSchema(SchemaDefinition.builder().withPojo(clazz).build());
     }
 
     /**
@@ -302,7 +302,7 @@ static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF_NAT
      */
     static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF_NATIVE(
             SchemaDefinition<T> schemaDefinition) {
-        return DefaultImplementation.newProtobufNativeSchema(schemaDefinition);
+        return DefaultImplementation.getDefaultImplementation().newProtobufNativeSchema(schemaDefinition);
     }
 
     /**
@@ -312,7 +312,7 @@ static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> PROTOBUF_NAT
      * @return a Schema instance
      */
     static <T> Schema<T> AVRO(Class<T> pojo) {
-        return DefaultImplementation.newAvroSchema(SchemaDefinition.builder().withPojo(pojo).build());
+        return DefaultImplementation.getDefaultImplementation().newAvroSchema(SchemaDefinition.builder().withPojo(pojo).build());
     }
 
     /**
@@ -322,7 +322,7 @@ static <T> Schema<T> AVRO(Class<T> pojo) {
      * @return a Schema instance
      */
     static <T> Schema<T> AVRO(SchemaDefinition<T> schemaDefinition) {
-        return DefaultImplementation.newAvroSchema(schemaDefinition);
+        return DefaultImplementation.getDefaultImplementation().newAvroSchema(schemaDefinition);
     }
 
     /**
@@ -332,7 +332,7 @@ static <T> Schema<T> AVRO(SchemaDefinition<T> schemaDefinition) {
      * @return a Schema instance
      */
     static <T> Schema<T> JSON(Class<T> pojo) {
-        return DefaultImplementation.newJSONSchema(SchemaDefinition.builder().withPojo(pojo).build());
+        return DefaultImplementation.getDefaultImplementation().newJSONSchema(SchemaDefinition.builder().withPojo(pojo).build());
     }
 
     /**
@@ -342,35 +342,35 @@ static <T> Schema<T> JSON(Class<T> pojo) {
      * @return a Schema instance
      */
     static <T> Schema<T> JSON(SchemaDefinition schemaDefinition) {
-        return DefaultImplementation.newJSONSchema(schemaDefinition);
+        return DefaultImplementation.getDefaultImplementation().newJSONSchema(schemaDefinition);
     }
 
     /**
      * Key Value Schema using passed in schema type, support JSON and AVRO currently.
      */
     static <K, V> Schema<KeyValue<K, V>> KeyValue(Class<K> key, Class<V> value, SchemaType type) {
-        return DefaultImplementation.newKeyValueSchema(key, value, type);
+        return DefaultImplementation.getDefaultImplementation().newKeyValueSchema(key, value, type);
     }
 
     /**
      * Schema that can be used to encode/decode KeyValue.
      */
     static Schema<KeyValue<byte[], byte[]>> KV_BYTES() {
-        return DefaultImplementation.newKeyValueBytesSchema();
+        return DefaultImplementation.getDefaultImplementation().newKeyValueBytesSchema();
     }
 
     /**
      * Key Value Schema whose underneath key and value schemas are JSONSchema.
      */
     static <K, V> Schema<KeyValue<K, V>> KeyValue(Class<K> key, Class<V> value) {
-        return DefaultImplementation.newKeyValueSchema(key, value, SchemaType.JSON);
+        return DefaultImplementation.getDefaultImplementation().newKeyValueSchema(key, value, SchemaType.JSON);
     }
 
     /**
      * Key Value Schema using passed in key and value schemas.
      */
     static <K, V> Schema<KeyValue<K, V>> KeyValue(Schema<K> key, Schema<V> value) {
-        return DefaultImplementation.newKeyValueSchema(key, value);
+        return DefaultImplementation.getDefaultImplementation().newKeyValueSchema(key, value);
     }
 
     /**
@@ -378,7 +378,7 @@ static <K, V> Schema<KeyValue<K, V>> KeyValue(Schema<K> key, Schema<V> value) {
      */
     static <K, V> Schema<KeyValue<K, V>> KeyValue(Schema<K> key, Schema<V> value,
         KeyValueEncodingType keyValueEncodingType) {
-        return DefaultImplementation.newKeyValueSchema(key, value, keyValueEncodingType);
+        return DefaultImplementation.getDefaultImplementation().newKeyValueSchema(key, value, keyValueEncodingType);
     }
 
     @Deprecated
@@ -396,7 +396,7 @@ static Schema<GenericRecord> AUTO() {
      * @return the auto schema instance
      */
     static Schema<GenericRecord> AUTO_CONSUME() {
-        return DefaultImplementation.newAutoConsumeSchema();
+        return DefaultImplementation.getDefaultImplementation().newAutoConsumeSchema();
     }
 
     /**
@@ -411,7 +411,7 @@ static Schema<GenericRecord> AUTO_CONSUME() {
      * @return the auto schema instance
      */
     static Schema<byte[]> AUTO_PRODUCE_BYTES() {
-        return DefaultImplementation.newAutoProduceSchema();
+        return DefaultImplementation.getDefaultImplementation().newAutoProduceSchema();
     }
 
     /**
@@ -423,7 +423,7 @@ static Schema<byte[]> AUTO_PRODUCE_BYTES() {
      * @see #AUTO_PRODUCE_BYTES()
      */
     static Schema<byte[]> AUTO_PRODUCE_BYTES(Schema<?> schema) {
-        return DefaultImplementation.newAutoProduceSchema(schema);
+        return DefaultImplementation.getDefaultImplementation().newAutoProduceSchema(schema);
     }
 
     /**
@@ -435,13 +435,13 @@ static Schema<byte[]> AUTO_PRODUCE_BYTES(Schema<?> schema) {
      * @since 2.9.0
      */
     static Schema<byte[]> NATIVE_AVRO(Object schema) {
-        return DefaultImplementation.newAutoProduceValidatedAvroSchema(schema);
+        return DefaultImplementation.getDefaultImplementation().newAutoProduceValidatedAvroSchema(schema);
     }
 
     // CHECKSTYLE.ON: MethodName
 
     static Schema<?> getSchema(SchemaInfo schemaInfo) {
-        return DefaultImplementation.getSchema(schemaInfo);
+        return DefaultImplementation.getDefaultImplementation().getSchema(schemaInfo);
     }
 
     /**
@@ -453,6 +453,6 @@ static Schema<?> getSchema(SchemaInfo schemaInfo) {
      * @return a generic schema instance
      */
     static GenericSchema<GenericRecord> generic(SchemaInfo schemaInfo) {
-        return DefaultImplementation.getGenericSchema(schemaInfo);
+        return DefaultImplementation.getDefaultImplementation().getGenericSchema(schemaInfo);
     }
 }
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaBuilder.java
index b9e820844761f..8448757bc96b9 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaBuilder.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaBuilder.java
@@ -36,7 +36,7 @@ public interface SchemaBuilder {
      * @return builder to build the schema for a record.
      */
     static RecordSchemaBuilder record(String name) {
-        return DefaultImplementation.newRecordSchemaBuilder(name);
+        return DefaultImplementation.getDefaultImplementation().newRecordSchemaBuilder(name);
     }
 
 }
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinition.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinition.java
index 720d1435e58a4..dffa5e421b202 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinition.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinition.java
@@ -20,6 +20,7 @@
 
 import java.util.Map;
 import java.util.Optional;
+
 import org.apache.pulsar.client.internal.DefaultImplementation;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
@@ -37,7 +38,7 @@ public interface SchemaDefinition<T> {
      * @return the {@link SchemaDefinition}
      */
     static <T> SchemaDefinitionBuilder<T> builder() {
-        return DefaultImplementation.newSchemaDefinitionBuilder();
+        return DefaultImplementation.getDefaultImplementation().newSchemaDefinitionBuilder();
     }
 
     /**
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
index a6020d67c567d..6fe1799f9dead 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
@@ -18,507 +18,31 @@
  */
 package org.apache.pulsar.client.internal;
 
-import static org.apache.pulsar.client.internal.ReflectionUtils.catchExceptions;
-import static org.apache.pulsar.client.internal.ReflectionUtils.getConstructor;
-import static org.apache.pulsar.client.internal.ReflectionUtils.getStaticMethod;
-import static org.apache.pulsar.client.internal.ReflectionUtils.newClassInstance;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.Method;
-import java.nio.ByteBuffer;
-import java.nio.charset.Charset;
-import java.sql.Time;
-import java.sql.Timestamp;
-import java.time.Instant;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.util.Date;
-import java.util.Map;
-import java.util.function.Supplier;
-import lombok.experimental.UtilityClass;
-import org.apache.pulsar.client.api.Authentication;
-import org.apache.pulsar.client.api.BatcherBuilder;
-import org.apache.pulsar.client.api.ClientBuilder;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
-import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.api.schema.GenericRecord;
-import org.apache.pulsar.client.api.schema.GenericSchema;
-import org.apache.pulsar.client.api.schema.RecordSchemaBuilder;
-import org.apache.pulsar.client.api.schema.SchemaDefinition;
-import org.apache.pulsar.client.api.schema.SchemaDefinitionBuilder;
-import org.apache.pulsar.common.schema.KeyValue;
-import org.apache.pulsar.common.schema.KeyValueEncodingType;
-import org.apache.pulsar.common.schema.SchemaInfo;
-import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
-import org.apache.pulsar.common.schema.SchemaType;
-
 /**
- * Helper class for class instantiations and it also contains methods to work with schemas.
+ * This class loads the implementation for {@link PulsarClientImplementationBinding}
+ * and allows you to decouple the API from the actual implementation.
+ * <b>This class is internal to the Pulsar API implementation, and it is not part of the public API
+ * it is not meant to be used by client applications.</b>
  */
-@SuppressWarnings("unchecked")
-@UtilityClass
 public class DefaultImplementation {
-
-    private static final Class<ClientBuilder> CLIENT_BUILDER_IMPL = newClassInstance(
-            "org.apache.pulsar.client.impl.ClientBuilderImpl");
-
-    private static final Constructor<MessageId> MESSAGE_ID_IMPL_long_long_int = getConstructor(
-            "org.apache.pulsar.client.impl.MessageIdImpl",
-            Long.TYPE, Long.TYPE, Integer.TYPE);
-
-    private static final Method MESSAGE_ID_IMPL_fromByteArray = getStaticMethod(
-            "org.apache.pulsar.client.impl.MessageIdImpl", "fromByteArray",
-            byte[].class);
-    private static final Method MESSAGE_ID_IMPL_fromByteArrayWithTopic = getStaticMethod(
-            "org.apache.pulsar.client.impl.MessageIdImpl",
-            "fromByteArrayWithTopic", byte[].class, String.class);
-
-    private static final Constructor<Authentication> AUTHENTICATION_TOKEN_String = getConstructor(
-            "org.apache.pulsar.client.impl.auth.AuthenticationToken", String.class);
-
-    private static final Constructor<Authentication> AUTHENTICATION_TOKEN_Supplier = getConstructor(
-            "org.apache.pulsar.client.impl.auth.AuthenticationToken", Supplier.class);
-
-    private static final Constructor<Authentication> AUTHENTICATION_TLS_String_String = getConstructor(
-            "org.apache.pulsar.client.impl.auth.AuthenticationTls", String.class, String.class);
-
-    private static final Constructor<SchemaDefinitionBuilder> SCHEMA_DEFINITION_BUILDER_CONSTRUCTOR = getConstructor(
-            "org.apache.pulsar.client.impl.schema.SchemaDefinitionBuilderImpl");
-
-    public static <T> SchemaDefinitionBuilder<T> newSchemaDefinitionBuilder() {
-        return catchExceptions(() -> (SchemaDefinitionBuilder<T>) SCHEMA_DEFINITION_BUILDER_CONSTRUCTOR.newInstance());
-    }
-
-    public static ClientBuilder newClientBuilder() {
-        return catchExceptions(() -> CLIENT_BUILDER_IMPL.newInstance());
-    }
-
-    public static MessageId newMessageId(long ledgerId, long entryId, int partitionIndex) {
-        return catchExceptions(() -> MESSAGE_ID_IMPL_long_long_int.newInstance(ledgerId, entryId, partitionIndex));
-    }
-
-    public static MessageId newMessageIdFromByteArray(byte[] data) {
-        return catchExceptions(() -> (MessageId) MESSAGE_ID_IMPL_fromByteArray.invoke(null, data));
-    }
-
-    public static MessageId newMessageIdFromByteArrayWithTopic(byte[] data, String topicName) {
-        return catchExceptions(() -> (MessageId) MESSAGE_ID_IMPL_fromByteArrayWithTopic.invoke(null, data, topicName));
-    }
-
-    public static Authentication newAuthenticationToken(String token) {
-        return catchExceptions(() -> AUTHENTICATION_TOKEN_String.newInstance(token));
-    }
-
-    public static Authentication newAuthenticationToken(Supplier<String> supplier) {
-        return catchExceptions(() -> AUTHENTICATION_TOKEN_Supplier.newInstance(supplier));
-    }
-
-    public static Authentication newAuthenticationTLS(String certFilePath, String keyFilePath) {
-        return catchExceptions(
-                () -> AUTHENTICATION_TLS_String_String.newInstance(certFilePath, keyFilePath));
-    }
-
-    public static Authentication createAuthentication(String authPluginClassName, String authParamsString)
-            throws UnsupportedAuthenticationException {
-        return catchExceptions(
-                () -> (Authentication) getStaticMethod("org.apache.pulsar.client.impl.AuthenticationUtil", "create",
-                        String.class, String.class)
-                                .invoke(null, authPluginClassName, authParamsString));
-    }
-
-    public static Authentication createAuthentication(String authPluginClassName, Map<String, String> authParams)
-            throws UnsupportedAuthenticationException {
-        return catchExceptions(
-                () -> (Authentication) getStaticMethod("org.apache.pulsar.client.impl.AuthenticationUtil", "create",
-                        String.class, Map.class)
-                                .invoke(null, authPluginClassName, authParams));
-    }
-
-    public static Schema<byte[]> newBytesSchema() {
-        return catchExceptions(
-                () -> (Schema<byte[]>) newClassInstance("org.apache.pulsar.client.impl.schema.BytesSchema")
-                        .newInstance());
-    }
-
-    public static Schema<String> newStringSchema() {
-        return catchExceptions(
-                () -> (Schema<String>) newClassInstance("org.apache.pulsar.client.impl.schema.StringSchema")
-                        .newInstance());
-    }
-
-    public static Schema<String> newStringSchema(Charset charset) {
-        return catchExceptions(
-                () -> (Schema<String>) getConstructor(
-                    "org.apache.pulsar.client.impl.schema.StringSchema", Charset.class)
-                        .newInstance(charset));
-    }
-
-    public static Schema<Byte> newByteSchema() {
-        return catchExceptions(
-                () -> (Schema<Byte>) newClassInstance("org.apache.pulsar.client.impl.schema.ByteSchema")
-                        .newInstance());
-    }
-
-    public static Schema<Short> newShortSchema() {
-        return catchExceptions(
-                () -> (Schema<Short>) newClassInstance("org.apache.pulsar.client.impl.schema.ShortSchema")
-                        .newInstance());
-    }
-
-    public static Schema<Integer> newIntSchema() {
-        return catchExceptions(
-                () -> (Schema<Integer>) newClassInstance("org.apache.pulsar.client.impl.schema.IntSchema")
-                        .newInstance());
-    }
-
-    public static Schema<Long> newLongSchema() {
-        return catchExceptions(
-                () -> (Schema<Long>) newClassInstance("org.apache.pulsar.client.impl.schema.LongSchema")
-                        .newInstance());
-    }
-
-    public static Schema<Boolean> newBooleanSchema() {
-        return catchExceptions(
-                () -> (Schema<Boolean>) newClassInstance("org.apache.pulsar.client.impl.schema.BooleanSchema")
-                        .newInstance());
-    }
-
-    public static Schema<ByteBuffer> newByteBufferSchema() {
-        return catchExceptions(
-                () -> (Schema<ByteBuffer>) newClassInstance("org.apache.pulsar.client.impl.schema.ByteBufferSchema")
-                        .newInstance());
-    }
-
-    public static Schema<Float> newFloatSchema() {
-        return catchExceptions(
-                () -> (Schema<Float>) newClassInstance("org.apache.pulsar.client.impl.schema.FloatSchema")
-                        .newInstance());
-    }
-
-    public static Schema<Double> newDoubleSchema() {
-        return catchExceptions(
-                () -> (Schema<Double>) newClassInstance("org.apache.pulsar.client.impl.schema.DoubleSchema")
-                        .newInstance());
-    }
-
-    public static Schema<Date> newDateSchema() {
-        return catchExceptions(
-                () -> (Schema<Date>) getStaticMethod(
-                    "org.apache.pulsar.client.impl.schema.DateSchema", "of", null)
-                        .invoke(null, null));
-    }
-
-    public static Schema<Time> newTimeSchema() {
-        return catchExceptions(
-              () -> (Schema<Time>) getStaticMethod(
-                  "org.apache.pulsar.client.impl.schema.TimeSchema", "of", null)
-                    .invoke(null, null));
-    }
-
-    public static Schema<Timestamp> newTimestampSchema() {
-        return catchExceptions(
-              () -> (Schema<Timestamp>) getStaticMethod(
-                  "org.apache.pulsar.client.impl.schema.TimestampSchema", "of", null)
-                    .invoke(null, null));
-    }
-    public static Schema<Instant> newInstantSchema() {
-        return catchExceptions(
-              () -> (Schema<Instant>) getStaticMethod(
-                  "org.apache.pulsar.client.impl.schema.InstantSchema", "of", null)
-                    .invoke(null, null));
-    }
-    public static Schema<LocalDate> newLocalDateSchema() {
-        return catchExceptions(
-              () -> (Schema<LocalDate>) getStaticMethod(
-                  "org.apache.pulsar.client.impl.schema.LocalDateSchema", "of", null)
-                    .invoke(null, null));
-    }
-    public static Schema<LocalTime> newLocalTimeSchema() {
-        return catchExceptions(
-              () -> (Schema<LocalTime>) getStaticMethod(
-                  "org.apache.pulsar.client.impl.schema.LocalTimeSchema", "of", null)
-                    .invoke(null, null));
-    }
-    public static Schema<LocalDateTime> newLocalDateTimeSchema() {
-        return catchExceptions(
-              () -> (Schema<LocalDateTime>) getStaticMethod(
-                  "org.apache.pulsar.client.impl.schema.LocalDateTimeSchema", "of", null)
-                    .invoke(null, null));
-    }
-
-    public static <T> Schema<T> newAvroSchema(SchemaDefinition schemaDefinition) {
-        return catchExceptions(
-                () -> (Schema<T>) getStaticMethod(
-                    "org.apache.pulsar.client.impl.schema.AvroSchema", "of", SchemaDefinition.class)
-                        .invoke(null, schemaDefinition));
-    }
-
-    public static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufSchema(
-        SchemaDefinition schemaDefinition) {
-        return catchExceptions(
-            () -> (Schema<T>) getStaticMethod(
-                "org.apache.pulsar.client.impl.schema.ProtobufSchema", "of", SchemaDefinition.class)
-                .invoke(null, schemaDefinition));
-    }
-
-    public static <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufNativeSchema(
-            SchemaDefinition schemaDefinition) {
-        return catchExceptions(
-                () -> (Schema<T>) getStaticMethod(
-                        "org.apache.pulsar.client.impl.schema.ProtobufNativeSchema", "of", SchemaDefinition.class)
-                        .invoke(null, schemaDefinition));
-    }
-
-    public static <T> Schema<T> newJSONSchema(SchemaDefinition schemaDefinition) {
-        return catchExceptions(
-                () -> (Schema<T>) getStaticMethod(
-                    "org.apache.pulsar.client.impl.schema.JSONSchema", "of", SchemaDefinition.class)
-                        .invoke(null, schemaDefinition));
-    }
-
-    public static Schema<GenericRecord> newAutoConsumeSchema() {
-        return catchExceptions(
-                () -> (Schema<GenericRecord>) newClassInstance(
-                    "org.apache.pulsar.client.impl.schema.AutoConsumeSchema")
-                        .newInstance());
-    }
-
-    public static Schema<byte[]> newAutoProduceSchema() {
-        return catchExceptions(
-                () -> (Schema<byte[]>) newClassInstance(
-                    "org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema")
-                        .newInstance());
-    }
-
-    public static Schema<byte[]> newAutoProduceSchema(Schema<?> schema) {
-        return catchExceptions(
-                () -> (Schema<byte[]>) getConstructor(
-                    "org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema", Schema.class)
-                        .newInstance(schema));
-    }
-
-    public static Schema<byte[]> newAutoProduceValidatedAvroSchema(Object schema) {
-        return catchExceptions(
-                () -> (Schema<byte[]>) getConstructor(
-                    "org.apache.pulsar.client.impl.schema.NativeAvroBytesSchema", Object.class)
-                        .newInstance(schema));
-    }
-
-    public static Schema<KeyValue<byte[], byte[]>> newKeyValueBytesSchema() {
-        return catchExceptions(
-                () -> (Schema<KeyValue<byte[], byte[]>>) getStaticMethod(
-                    "org.apache.pulsar.client.impl.schema.KeyValueSchemaImpl",
-                        "kvBytes").invoke(null));
-    }
-
-    public static <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Schema<K> keySchema, Schema<V> valueSchema) {
-        return catchExceptions(
-                () -> (Schema<KeyValue<K, V>>) getStaticMethod(
-                    "org.apache.pulsar.client.impl.schema.KeyValueSchemaImpl",
-                        "of", Schema.class, Schema.class).invoke(null, keySchema, valueSchema));
-    }
-
-    public static <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Schema<K> keySchema, Schema<V> valueSchema,
-                                                                  KeyValueEncodingType keyValueEncodingType) {
-        return catchExceptions(
-                () -> (Schema<KeyValue<K, V>>) getStaticMethod("org.apache.pulsar.client.impl.schema.KeyValueSchemaImpl",
-                        "of", Schema.class, Schema.class, KeyValueEncodingType.class)
-                        .invoke(null, keySchema, valueSchema, keyValueEncodingType));
-    }
-
-    public static <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Class<K> key, Class<V> value, SchemaType type) {
-        return catchExceptions(
-                () -> (Schema<KeyValue<K, V>>) getStaticMethod("org.apache.pulsar.client.impl.schema.KeyValueSchemaImpl",
-                        "of", Class.class, Class.class, SchemaType.class).invoke(null, key, value, type));
-    }
-
-    public static Schema<?> getSchema(SchemaInfo schemaInfo) {
-        return catchExceptions(
-                () -> (Schema<?>) getStaticMethod("org.apache.pulsar.client.impl.schema.AutoConsumeSchema",
-                        "getSchema", SchemaInfo.class).invoke(null, schemaInfo));
-    }
-
-    public static GenericSchema<GenericRecord> getGenericSchema(SchemaInfo schemaInfo) {
-        switch (schemaInfo.getType()) {
-            case PROTOBUF_NATIVE:
-                return (GenericSchema) ReflectionUtils.catchExceptions(() -> {
-                    return (GenericSchema) ReflectionUtils.getStaticMethod(
-                            "org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema",
-                            "of", new Class[]{SchemaInfo.class}).invoke((Object) null, schemaInfo);
-                });
-            default:
-                return (GenericSchema) ReflectionUtils.catchExceptions(() -> {
-                    return (GenericSchema) ReflectionUtils.getStaticMethod(
-                            "org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl",
-                            "of", new Class[]{SchemaInfo.class}).invoke((Object) null, schemaInfo);
-
-                });
+    private static final PulsarClientImplementationBinding IMPLEMENTATION;
+    static {
+        PulsarClientImplementationBinding impl = null;
+        try {
+            impl = (PulsarClientImplementationBinding) ReflectionUtils.newClassInstance("org.apache.pulsar.client.impl.PulsarClientImplementationBindingImpl")
+                    .getConstructor().newInstance();
+        } catch (Throwable error) {
+            throw new RuntimeException("Cannot load Pulsar Client Implementation: "+error, error);
         }
-
-    }
-
-    public static RecordSchemaBuilder newRecordSchemaBuilder(String name) {
-        return catchExceptions(
-                () -> (RecordSchemaBuilder) getConstructor(
-                    "org.apache.pulsar.client.impl.schema.RecordSchemaBuilderImpl",
-                    String.class).newInstance(name));
-    }
-
-    /**
-     * Decode the kv encoding type from the schema info.
-     *
-     * @param schemaInfo the schema info
-     * @return the kv encoding type
-     */
-    public static KeyValueEncodingType decodeKeyValueEncodingType(SchemaInfo schemaInfo) {
-        return catchExceptions(
-            () -> (KeyValueEncodingType) getStaticMethod("org.apache.pulsar.client.impl.schema.KeyValueSchemaInfo",
-                "decodeKeyValueEncodingType", SchemaInfo.class
-            ).invoke(null, schemaInfo));
-    }
-
-    /**
-     * Encode key & value into schema into a KeyValue schema.
-     *
-     * @param keySchema the key schema
-     * @param valueSchema the value schema
-     * @param keyValueEncodingType the encoding type to encode and decode key value pair
-     * @return the final schema info
-     */
-    public static <K, V> SchemaInfo encodeKeyValueSchemaInfo(Schema<K> keySchema,
-                                                             Schema<V> valueSchema,
-                                                             KeyValueEncodingType keyValueEncodingType) {
-        return encodeKeyValueSchemaInfo("KeyValue", keySchema, valueSchema, keyValueEncodingType);
+        IMPLEMENTATION = impl;
     }
 
     /**
-     * Encode key & value into schema into a KeyValue schema.
-     *
-     * @param schemaName the final schema name
-     * @param keySchema the key schema
-     * @param valueSchema the value schema
-     * @param keyValueEncodingType the encoding type to encode and decode key value pair
-     * @return the final schema info
+     * Access the actual implementation of the Pulsar Client API
+     * @return the loaded implementation.
      */
-    public static <K, V> SchemaInfo encodeKeyValueSchemaInfo(String schemaName,
-                                                             Schema<K> keySchema,
-                                                             Schema<V> valueSchema,
-                                                             KeyValueEncodingType keyValueEncodingType) {
-        return catchExceptions(
-            () -> (SchemaInfo) getStaticMethod("org.apache.pulsar.client.impl.schema.KeyValueSchemaInfo",
-                "encodeKeyValueSchemaInfo", String.class, Schema.class, Schema.class, KeyValueEncodingType.class
-            ).invoke(null, schemaName, keySchema, valueSchema, keyValueEncodingType));
+    public static PulsarClientImplementationBinding getDefaultImplementation() {
+        return IMPLEMENTATION;
     }
 
-    /**
-     * Decode the key/value schema info to get key schema info and value schema info.
-     *
-     * @param schemaInfo key/value schema info.
-     * @return the pair of key schema info and value schema info
-     */
-    public static KeyValue<SchemaInfo, SchemaInfo> decodeKeyValueSchemaInfo(SchemaInfo schemaInfo) {
-        return catchExceptions(
-            () -> (KeyValue<SchemaInfo, SchemaInfo>) getStaticMethod(
-                "org.apache.pulsar.client.impl.schema.KeyValueSchemaInfo",
-                "decodeKeyValueSchemaInfo", SchemaInfo.class
-            ).invoke(null, schemaInfo));
-    }
-
-    /**
-     * Jsonify the schema info.
-     *
-     * @param schemaInfo the schema info
-     * @return the jsonified schema info
-     */
-    public static String jsonifySchemaInfo(SchemaInfo schemaInfo) {
-        return catchExceptions(
-            () -> (String) getStaticMethod("org.apache.pulsar.client.impl.schema.SchemaUtils",
-                "jsonifySchemaInfo", SchemaInfo.class
-            ).invoke(null, schemaInfo));
-    }
-
-    /**
-     * Jsonify the schema info with version.
-     *
-     * @param schemaInfoWithVersion the schema info with version
-     * @return the jsonified schema info with version
-     */
-    public static String jsonifySchemaInfoWithVersion(SchemaInfoWithVersion schemaInfoWithVersion) {
-        return catchExceptions(
-                () -> (String) getStaticMethod("org.apache.pulsar.client.impl.schema.SchemaUtils",
-                        "jsonifySchemaInfoWithVersion", SchemaInfoWithVersion.class
-                ).invoke(null, schemaInfoWithVersion));
-    }
-
-    /**
-     * Jsonify the key/value schema info.
-     *
-     * @param kvSchemaInfo the key/value schema info
-     * @return the jsonified schema info
-     */
-    public static String jsonifyKeyValueSchemaInfo(KeyValue<SchemaInfo, SchemaInfo> kvSchemaInfo) {
-        return catchExceptions(
-            () -> (String) getStaticMethod("org.apache.pulsar.client.impl.schema.SchemaUtils",
-                "jsonifyKeyValueSchemaInfo", KeyValue.class
-            ).invoke(null, kvSchemaInfo));
-    }
-
-    /**
-     * Convert the key/value schema data.
-     *
-     * @param kvSchemaInfo the key/value schema info
-     * @return the convert key/value schema data string
-     */
-    public static String convertKeyValueSchemaInfoDataToString(KeyValue<SchemaInfo, SchemaInfo> kvSchemaInfo) {
-        return catchExceptions(
-                () -> (String) getStaticMethod("org.apache.pulsar.client.impl.schema.SchemaUtils",
-                        "convertKeyValueSchemaInfoDataToString", KeyValue.class
-                ).invoke(null, kvSchemaInfo));
-    }
-
-    /**
-     * Convert the key/value schema info data json bytes to key/value schema info data bytes.
-     *
-     * @param keyValueSchemaInfoDataJsonBytes the key/value schema info data json bytes
-     * @return the key/value schema info data bytes
-     */
-    public static byte[] convertKeyValueDataStringToSchemaInfoSchema(byte[] keyValueSchemaInfoDataJsonBytes) {
-        return catchExceptions(
-                () -> (byte[]) getStaticMethod("org.apache.pulsar.client.impl.schema.SchemaUtils",
-                        "convertKeyValueDataStringToSchemaInfoSchema", byte[].class
-                ).invoke(null, keyValueSchemaInfoDataJsonBytes));
-    }
-
-    public static BatcherBuilder newDefaultBatcherBuilder() {
-        return catchExceptions(
-            () -> (BatcherBuilder) getConstructor("org.apache.pulsar.client.impl.DefaultBatcherBuilder")
-                    .newInstance());
-    }
-
-    public static BatcherBuilder newKeyBasedBatcherBuilder() {
-        return catchExceptions(
-                () -> (BatcherBuilder) getConstructor("org.apache.pulsar.client.impl.KeyBasedBatcherBuilder")
-                        .newInstance());
-    }
-
-    /**
-     * Retrieves ByteBuffer data into byte[].
-     * 
-     * @param byteBuffer
-     * @return
-     */
-    public static byte[] getBytes(ByteBuffer byteBuffer) {
-        if (byteBuffer == null) {
-            return null;
-        }
-        if (byteBuffer.hasArray() && byteBuffer.arrayOffset() == 0
-                && byteBuffer.array().length == byteBuffer.remaining()) {
-            return byteBuffer.array();
-        }
-        // Direct buffer is not backed by array and it needs to be read from direct memory
-        byte[] array = new byte[byteBuffer.remaining()];
-        byteBuffer.get(array);
-        return array;
-    }
 }
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/PulsarClientImplementationBinding.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/PulsarClientImplementationBinding.java
new file mode 100644
index 0000000000000..98b4d5c40c02a
--- /dev/null
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/PulsarClientImplementationBinding.java
@@ -0,0 +1,251 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.internal;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.BatcherBuilder;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.RecordSchemaBuilder;
+import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.api.schema.SchemaDefinitionBuilder;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
+import org.apache.pulsar.common.schema.SchemaType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.Date;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/**
+ * Helper class for class instantiations and it also contains methods to work with schemas.
+ * This interface allows you to not depend on the Implementation classes directly.
+ * The actual implementation of this class is loaded from {@link DefaultImplementation}.
+ */
+public interface PulsarClientImplementationBinding {
+    
+    <T> SchemaDefinitionBuilder<T> newSchemaDefinitionBuilder();
+
+    ClientBuilder newClientBuilder();
+
+    MessageId newMessageId(long ledgerId, long entryId, int partitionIndex);
+
+    MessageId newMessageIdFromByteArray(byte[] data) throws IOException;
+
+    MessageId newMessageIdFromByteArrayWithTopic(byte[] data, String topicName) throws IOException;
+
+    Authentication newAuthenticationToken(String token);
+
+    Authentication newAuthenticationToken(Supplier<String> supplier);
+
+    Authentication newAuthenticationTLS(String certFilePath, String keyFilePath);
+
+    Authentication createAuthentication(String authPluginClassName, String authParamsString)
+            throws PulsarClientException.UnsupportedAuthenticationException;
+
+    Authentication createAuthentication(String authPluginClassName, Map<String, String> authParams)
+            throws PulsarClientException.UnsupportedAuthenticationException;
+
+    Schema<byte[]> newBytesSchema();
+
+    Schema<String> newStringSchema();
+
+    Schema<String> newStringSchema(Charset charset);
+
+    Schema<Byte> newByteSchema();
+
+    Schema<Short> newShortSchema();
+
+    Schema<Integer> newIntSchema();
+
+    Schema<Long> newLongSchema();
+
+    Schema<Boolean> newBooleanSchema();
+
+    Schema<ByteBuffer> newByteBufferSchema();
+
+    Schema<Float> newFloatSchema();
+
+    Schema<Double> newDoubleSchema();
+
+    Schema<Date> newDateSchema();
+
+    Schema<Time> newTimeSchema();
+
+    Schema<Timestamp> newTimestampSchema();
+
+    Schema<Instant> newInstantSchema();
+
+    Schema<LocalDate> newLocalDateSchema();
+
+    Schema<LocalTime> newLocalTimeSchema();
+
+    Schema<LocalDateTime> newLocalDateTimeSchema();
+
+    <T> Schema<T> newAvroSchema(SchemaDefinition schemaDefinition);
+
+    <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufSchema(SchemaDefinition schemaDefinition);
+
+    <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufNativeSchema(SchemaDefinition schemaDefinition);
+
+    <T> Schema<T> newJSONSchema(SchemaDefinition schemaDefinition);
+
+    Schema<GenericRecord> newAutoConsumeSchema();
+
+    Schema<byte[]> newAutoProduceSchema();
+
+    Schema<byte[]> newAutoProduceSchema(Schema<?> schema);
+
+    Schema<byte[]> newAutoProduceValidatedAvroSchema(Object schema);
+
+    Schema<KeyValue<byte[], byte[]>> newKeyValueBytesSchema();
+
+    <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Schema<K> keySchema, Schema<V> valueSchema);
+
+    <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Schema<K> keySchema, Schema<V> valueSchema,
+                                                           KeyValueEncodingType keyValueEncodingType);
+
+    <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Class<K> key, Class<V> value, SchemaType type);
+
+    Schema<?> getSchema(SchemaInfo schemaInfo);
+
+    GenericSchema<GenericRecord> getGenericSchema(SchemaInfo schemaInfo);
+
+    RecordSchemaBuilder newRecordSchemaBuilder(String name);
+
+    /**
+     * Decode the kv encoding type from the schema info.
+     *
+     * @param schemaInfo the schema info
+     * @return the kv encoding type
+     */
+    KeyValueEncodingType decodeKeyValueEncodingType(SchemaInfo schemaInfo);
+
+    /**
+     * Encode key & value into schema into a KeyValue schema.
+     *
+     * @param keySchema            the key schema
+     * @param valueSchema          the value schema
+     * @param keyValueEncodingType the encoding type to encode and decode key value pair
+     * @return the final schema info
+     */
+    <K, V> SchemaInfo encodeKeyValueSchemaInfo(Schema<K> keySchema,
+                                                      Schema<V> valueSchema,
+                                                      KeyValueEncodingType keyValueEncodingType);
+
+    /**
+     * Encode key & value into schema into a KeyValue schema.
+     *
+     * @param schemaName           the final schema name
+     * @param keySchema            the key schema
+     * @param valueSchema          the value schema
+     * @param keyValueEncodingType the encoding type to encode and decode key value pair
+     * @return the final schema info
+     */
+    <K, V> SchemaInfo encodeKeyValueSchemaInfo(String schemaName,
+                                                      Schema<K> keySchema,
+                                                      Schema<V> valueSchema,
+                                                      KeyValueEncodingType keyValueEncodingType);
+
+    /**
+     * Decode the key/value schema info to get key schema info and value schema info.
+     *
+     * @param schemaInfo key/value schema info.
+     * @return the pair of key schema info and value schema info
+     */
+    KeyValue<SchemaInfo, SchemaInfo> decodeKeyValueSchemaInfo(SchemaInfo schemaInfo);
+
+    /**
+     * Jsonify the schema info.
+     *
+     * @param schemaInfo the schema info
+     * @return the jsonified schema info
+     */
+    String jsonifySchemaInfo(SchemaInfo schemaInfo);
+
+    /**
+     * Jsonify the schema info with version.
+     *
+     * @param schemaInfoWithVersion the schema info with version
+     * @return the jsonified schema info with version
+     */
+    String jsonifySchemaInfoWithVersion(SchemaInfoWithVersion schemaInfoWithVersion);
+
+    /**
+     * Jsonify the key/value schema info.
+     *
+     * @param kvSchemaInfo the key/value schema info
+     * @return the jsonified schema info
+     */
+    String jsonifyKeyValueSchemaInfo(KeyValue<SchemaInfo, SchemaInfo> kvSchemaInfo);
+
+    /**
+     * Convert the key/value schema data.
+     *
+     * @param kvSchemaInfo the key/value schema info
+     * @return the convert key/value schema data string
+     */
+    String convertKeyValueSchemaInfoDataToString(KeyValue<SchemaInfo, SchemaInfo> kvSchemaInfo) throws IOException;
+
+    /**
+     * Convert the key/value schema info data json bytes to key/value schema info data bytes.
+     *
+     * @param keyValueSchemaInfoDataJsonBytes the key/value schema info data json bytes
+     * @return the key/value schema info data bytes
+     */
+    byte[] convertKeyValueDataStringToSchemaInfoSchema(byte[] keyValueSchemaInfoDataJsonBytes) throws IOException;
+
+    BatcherBuilder newDefaultBatcherBuilder();
+
+    BatcherBuilder newKeyBasedBatcherBuilder();
+
+    /**
+     * Retrieves ByteBuffer data into byte[].
+     *
+     * @param byteBuffer
+     * @return
+     */
+    static byte[] getBytes(ByteBuffer byteBuffer) {
+        if (byteBuffer == null) {
+            return null;
+        }
+        if (byteBuffer.hasArray() && byteBuffer.arrayOffset() == 0
+                && byteBuffer.array().length == byteBuffer.remaining()) {
+            return byteBuffer.array();
+        }
+        // Direct buffer is not backed by array and it needs to be read from direct memory
+        byte[] array = new byte[byteBuffer.remaining()];
+        byteBuffer.get(array);
+        return array;
+    }
+}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfo.java b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfo.java
index 0070c4cbaa4bf..01ba7465c112b 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfo.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfo.java
@@ -19,7 +19,7 @@
 package org.apache.pulsar.common.schema;
 
 import java.util.Map;
-import org.apache.pulsar.client.internal.DefaultImplementation;
+
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
 
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfoWithVersion.java b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfoWithVersion.java
index dba56c59d6ecb..8ee742ddd8bbb 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfoWithVersion.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/SchemaInfoWithVersion.java
@@ -45,6 +45,6 @@ public class SchemaInfoWithVersion {
 
     @Override
     public String toString(){
-        return DefaultImplementation.jsonifySchemaInfoWithVersion(this);
+        return DefaultImplementation.getDefaultImplementation().jsonifySchemaInfoWithVersion(this);
     }
 }
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java
index aaac894395061..d41202c1ecac5 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java
@@ -19,7 +19,8 @@
 package org.apache.pulsar.client.cli;
 
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
-import static org.apache.pulsar.client.internal.DefaultImplementation.getBytes;
+import static org.apache.pulsar.client.internal.PulsarClientImplementationBinding.getBytes;
+
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import com.beust.jcommander.Parameters;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImplementationBindingImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImplementationBindingImpl.java
new file mode 100644
index 0000000000000..225e2467b416c
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImplementationBindingImpl.java
@@ -0,0 +1,382 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.Date;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.BatcherBuilder;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
+import org.apache.pulsar.client.api.schema.RecordSchemaBuilder;
+import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.api.schema.SchemaDefinitionBuilder;
+import org.apache.pulsar.client.impl.auth.AuthenticationTls;
+import org.apache.pulsar.client.impl.auth.AuthenticationToken;
+import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
+import org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema;
+import org.apache.pulsar.client.impl.schema.AvroSchema;
+import org.apache.pulsar.client.impl.schema.BooleanSchema;
+import org.apache.pulsar.client.impl.schema.ByteBufferSchema;
+import org.apache.pulsar.client.impl.schema.ByteSchema;
+import org.apache.pulsar.client.impl.schema.BytesSchema;
+import org.apache.pulsar.client.impl.schema.DateSchema;
+import org.apache.pulsar.client.impl.schema.DoubleSchema;
+import org.apache.pulsar.client.impl.schema.FloatSchema;
+import org.apache.pulsar.client.impl.schema.InstantSchema;
+import org.apache.pulsar.client.impl.schema.IntSchema;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.client.impl.schema.KeyValueSchemaImpl;
+import org.apache.pulsar.client.impl.schema.KeyValueSchemaInfo;
+import org.apache.pulsar.client.impl.schema.LocalDateSchema;
+import org.apache.pulsar.client.impl.schema.LocalDateTimeSchema;
+import org.apache.pulsar.client.impl.schema.LocalTimeSchema;
+import org.apache.pulsar.client.impl.schema.LongSchema;
+import org.apache.pulsar.client.impl.schema.NativeAvroBytesSchema;
+import org.apache.pulsar.client.impl.schema.ProtobufNativeSchema;
+import org.apache.pulsar.client.impl.schema.ProtobufSchema;
+import org.apache.pulsar.client.impl.schema.RecordSchemaBuilderImpl;
+import org.apache.pulsar.client.impl.schema.SchemaDefinitionBuilderImpl;
+import org.apache.pulsar.client.impl.schema.SchemaUtils;
+import org.apache.pulsar.client.impl.schema.ShortSchema;
+import org.apache.pulsar.client.impl.schema.StringSchema;
+import org.apache.pulsar.client.impl.schema.TimeSchema;
+import org.apache.pulsar.client.impl.schema.TimestampSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
+import org.apache.pulsar.client.internal.PulsarClientImplementationBinding;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
+import org.apache.pulsar.common.schema.SchemaType;
+
+/**
+ * Helper class for class instantiations and it also contains methods to work with schemas.
+ */
+@SuppressWarnings("unchecked")
+public final class PulsarClientImplementationBindingImpl implements PulsarClientImplementationBinding {
+
+    public <T> SchemaDefinitionBuilder<T> newSchemaDefinitionBuilder() {
+        return new SchemaDefinitionBuilderImpl();
+    }
+
+    public ClientBuilder newClientBuilder() {
+        return new ClientBuilderImpl();
+    }
+
+    public MessageId newMessageId(long ledgerId, long entryId, int partitionIndex) {
+        return new MessageIdImpl(ledgerId, entryId, partitionIndex);
+    }
+
+    public MessageId newMessageIdFromByteArray(byte[] data) throws IOException {
+        return MessageIdImpl.fromByteArray(data);
+    }
+
+    public MessageId newMessageIdFromByteArrayWithTopic(byte[] data, String topicName) throws IOException {
+        return MessageIdImpl.fromByteArrayWithTopic(data, topicName);
+    }
+
+    public Authentication newAuthenticationToken(String token) {
+        return new AuthenticationToken(token);
+    }
+
+    public Authentication newAuthenticationToken(Supplier<String> supplier) {
+        return new AuthenticationToken(supplier);
+    }
+
+    public Authentication newAuthenticationTLS(String certFilePath, String keyFilePath) {
+        return new AuthenticationTls(certFilePath, keyFilePath);
+    }
+
+    public Authentication createAuthentication(String authPluginClassName, String authParamsString)
+            throws PulsarClientException.UnsupportedAuthenticationException {
+        return AuthenticationUtil.create(authPluginClassName, authParamsString);
+    }
+
+    public Authentication createAuthentication(String authPluginClassName, Map<String, String> authParams)
+            throws PulsarClientException.UnsupportedAuthenticationException {
+        return AuthenticationUtil.create(authPluginClassName, authParams);
+    }
+
+    public Schema<byte[]> newBytesSchema() {
+        return new BytesSchema();
+    }
+
+    public Schema<String> newStringSchema() {
+        return new StringSchema();
+    }
+
+    public Schema<String> newStringSchema(Charset charset) {
+        return new StringSchema(charset);
+    }
+
+    public Schema<Byte> newByteSchema() {
+        return new ByteSchema();
+    }
+
+    public Schema<Short> newShortSchema() {
+        return new ShortSchema();
+    }
+
+    public Schema<Integer> newIntSchema() {
+        return new IntSchema();
+    }
+
+    public Schema<Long> newLongSchema() {
+        return new LongSchema();
+    }
+
+    public Schema<Boolean> newBooleanSchema() {
+        return new BooleanSchema();
+    }
+
+    public Schema<ByteBuffer> newByteBufferSchema() {
+        return new ByteBufferSchema();
+    }
+
+    public Schema<Float> newFloatSchema() {
+        return new FloatSchema();
+    }
+
+    public Schema<Double> newDoubleSchema() {
+        return new DoubleSchema();
+    }
+
+    public Schema<Date> newDateSchema() {
+        return DateSchema.of();
+    }
+
+    public Schema<Time> newTimeSchema() {
+        return TimeSchema.of();
+    }
+
+    public Schema<Timestamp> newTimestampSchema() {
+        return TimestampSchema.of();
+    }
+
+    public Schema<Instant> newInstantSchema() {
+        return InstantSchema.of();
+    }
+
+    public Schema<LocalDate> newLocalDateSchema() {
+        return LocalDateSchema.of();
+    }
+
+    public Schema<LocalTime> newLocalTimeSchema() {
+        return LocalTimeSchema.of();
+    }
+
+    public Schema<LocalDateTime> newLocalDateTimeSchema() {
+        return LocalDateTimeSchema.of();
+    }
+
+    public <T> Schema<T> newAvroSchema(SchemaDefinition schemaDefinition) {
+        return AvroSchema.of(schemaDefinition);
+    }
+
+    public <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufSchema(
+            SchemaDefinition schemaDefinition) {
+        return ProtobufSchema.of(schemaDefinition);
+    }
+
+    public <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufNativeSchema(
+            SchemaDefinition schemaDefinition) {
+        return ProtobufNativeSchema.of(schemaDefinition);
+    }
+
+    public <T> Schema<T> newJSONSchema(SchemaDefinition schemaDefinition) {
+        return JSONSchema.of(schemaDefinition);
+    }
+
+    public Schema<GenericRecord> newAutoConsumeSchema() {
+        return new AutoConsumeSchema();
+    }
+
+    public Schema<byte[]> newAutoProduceSchema() {
+        return new AutoProduceBytesSchema();
+    }
+
+    public Schema<byte[]> newAutoProduceSchema(Schema<?> schema) {
+        return new AutoProduceBytesSchema(schema);
+    }
+
+    public Schema<byte[]> newAutoProduceValidatedAvroSchema(Object schema) {
+        return new NativeAvroBytesSchema(schema);
+    }
+
+    public Schema<KeyValue<byte[], byte[]>> newKeyValueBytesSchema() {
+        return KeyValueSchemaImpl.kvBytes();
+    }
+
+    public <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Schema<K> keySchema, Schema<V> valueSchema) {
+        return KeyValueSchemaImpl.of(keySchema, valueSchema);
+    }
+
+    public <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Schema<K> keySchema, Schema<V> valueSchema,
+                                                    KeyValueEncodingType keyValueEncodingType) {
+        return KeyValueSchemaImpl.of(keySchema, valueSchema, keyValueEncodingType);
+    }
+
+    public <K, V> Schema<KeyValue<K, V>> newKeyValueSchema(Class<K> key, Class<V> value, SchemaType type) {
+        return KeyValueSchemaImpl.of(key, value, type);
+    }
+
+    public Schema<?> getSchema(SchemaInfo schemaInfo) {
+        return AutoConsumeSchema.getSchema(schemaInfo);
+    }
+
+    public GenericSchema<GenericRecord> getGenericSchema(SchemaInfo schemaInfo) {
+        switch (schemaInfo.getType()) {
+            case PROTOBUF_NATIVE:
+                return GenericProtobufNativeSchema.of(schemaInfo);
+            default:
+                return GenericSchemaImpl.of(schemaInfo);
+        }
+
+    }
+
+    public RecordSchemaBuilder newRecordSchemaBuilder(String name) {
+        return new RecordSchemaBuilderImpl(name);
+    }
+
+    /**
+     * Decode the kv encoding type from the schema info.
+     *
+     * @param schemaInfo the schema info
+     * @return the kv encoding type
+     */
+    public KeyValueEncodingType decodeKeyValueEncodingType(SchemaInfo schemaInfo) {
+        return KeyValueSchemaInfo.decodeKeyValueEncodingType(schemaInfo);
+    }
+
+    /**
+     * Encode key & value into schema into a KeyValue schema.
+     *
+     * @param keySchema            the key schema
+     * @param valueSchema          the value schema
+     * @param keyValueEncodingType the encoding type to encode and decode key value pair
+     * @return the final schema info
+     */
+    public <K, V> SchemaInfo encodeKeyValueSchemaInfo(Schema<K> keySchema,
+                                               Schema<V> valueSchema,
+                                               KeyValueEncodingType keyValueEncodingType) {
+        return encodeKeyValueSchemaInfo("KeyValue", keySchema, valueSchema, keyValueEncodingType);
+    }
+
+    /**
+     * Encode key & value into schema into a KeyValue schema.
+     *
+     * @param schemaName           the final schema name
+     * @param keySchema            the key schema
+     * @param valueSchema          the value schema
+     * @param keyValueEncodingType the encoding type to encode and decode key value pair
+     * @return the final schema info
+     */
+    public <K, V> SchemaInfo encodeKeyValueSchemaInfo(String schemaName,
+                                               Schema<K> keySchema,
+                                               Schema<V> valueSchema,
+                                               KeyValueEncodingType keyValueEncodingType) {
+        return KeyValueSchemaInfo.encodeKeyValueSchemaInfo(schemaName, keySchema, valueSchema, keyValueEncodingType);
+    }
+
+    /**
+     * Decode the key/value schema info to get key schema info and value schema info.
+     *
+     * @param schemaInfo key/value schema info.
+     * @return the pair of key schema info and value schema info
+     */
+    public KeyValue<SchemaInfo, SchemaInfo> decodeKeyValueSchemaInfo(SchemaInfo schemaInfo) {
+        return KeyValueSchemaInfo.decodeKeyValueSchemaInfo(schemaInfo);
+    }
+
+    /**
+     * Jsonify the schema info.
+     *
+     * @param schemaInfo the schema info
+     * @return the jsonified schema info
+     */
+    public String jsonifySchemaInfo(SchemaInfo schemaInfo) {
+        return SchemaUtils.jsonifySchemaInfo(schemaInfo);
+    }
+
+    /**
+     * Jsonify the schema info with version.
+     *
+     * @param schemaInfoWithVersion the schema info with version
+     * @return the jsonified schema info with version
+     */
+    public String jsonifySchemaInfoWithVersion(SchemaInfoWithVersion schemaInfoWithVersion) {
+        return SchemaUtils.jsonifySchemaInfoWithVersion(schemaInfoWithVersion);
+    }
+
+    /**
+     * Jsonify the key/value schema info.
+     *
+     * @param kvSchemaInfo the key/value schema info
+     * @return the jsonified schema info
+     */
+    public String jsonifyKeyValueSchemaInfo(KeyValue<SchemaInfo, SchemaInfo> kvSchemaInfo) {
+        return SchemaUtils.jsonifyKeyValueSchemaInfo(kvSchemaInfo);
+    }
+
+    /**
+     * Convert the key/value schema data.
+     *
+     * @param kvSchemaInfo the key/value schema info
+     * @return the convert key/value schema data string
+     */
+    public String convertKeyValueSchemaInfoDataToString(KeyValue<SchemaInfo, SchemaInfo> kvSchemaInfo) throws IOException {
+        return SchemaUtils.convertKeyValueSchemaInfoDataToString(kvSchemaInfo);
+    }
+
+    /**
+     * Convert the key/value schema info data json bytes to key/value schema info data bytes.
+     *
+     * @param keyValueSchemaInfoDataJsonBytes the key/value schema info data json bytes
+     * @return the key/value schema info data bytes
+     */
+    public byte[] convertKeyValueDataStringToSchemaInfoSchema(byte[] keyValueSchemaInfoDataJsonBytes) throws IOException {
+        return SchemaUtils.convertKeyValueDataStringToSchemaInfoSchema(keyValueSchemaInfoDataJsonBytes);
+    }
+
+    public BatcherBuilder newDefaultBatcherBuilder() {
+        return new DefaultBatcherBuilder();
+    }
+
+    public BatcherBuilder newKeyBasedBatcherBuilder() {
+        return new KeyBasedBatcherBuilder();
+    }
+
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaInfoTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaInfoTest.java
index 9474b801bcdb9..b8bd72e8d07e9 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaInfoTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaInfoTest.java
@@ -33,6 +33,7 @@
 import org.apache.pulsar.client.api.schema.SchemaDefinition;
 import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Bar;
 import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Foo;
+import org.apache.pulsar.client.impl.PulsarClientImplementationBindingImpl;
 import org.apache.pulsar.client.internal.DefaultImplementation;
 import org.apache.pulsar.common.schema.KeyValue;
 import org.apache.pulsar.common.schema.KeyValueEncodingType;
@@ -88,7 +89,7 @@ public class KeyValueSchemaInfoTest {
 
     @Test(expectedExceptions = IllegalArgumentException.class)
     public void testDecodeNonKeyValueSchemaInfo() {
-        DefaultImplementation.decodeKeyValueSchemaInfo(
+        DefaultImplementation.getDefaultImplementation().decodeKeyValueSchemaInfo(
             FOO_SCHEMA.getSchemaInfo()
         );
     }
@@ -110,18 +111,18 @@ public void encodeDecodeKeyValueSchemaInfo(KeyValueEncodingType encodingType) {
         );
         SchemaInfo kvSchemaInfo = kvSchema.getSchemaInfo();
         assertEquals(
-            DefaultImplementation.decodeKeyValueEncodingType(kvSchemaInfo),
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueEncodingType(kvSchemaInfo),
             encodingType);
 
         SchemaInfo encodedSchemaInfo =
-            DefaultImplementation.encodeKeyValueSchemaInfo(FOO_SCHEMA, BAR_SCHEMA, encodingType);
+                DefaultImplementation.getDefaultImplementation().encodeKeyValueSchemaInfo(FOO_SCHEMA, BAR_SCHEMA, encodingType);
         assertEquals(encodedSchemaInfo, kvSchemaInfo);
         assertEquals(
-            DefaultImplementation.decodeKeyValueEncodingType(encodedSchemaInfo),
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueEncodingType(encodedSchemaInfo),
             encodingType);
 
         KeyValue<SchemaInfo, SchemaInfo> schemaInfoKeyValue =
-            DefaultImplementation.decodeKeyValueSchemaInfo(kvSchemaInfo);
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueSchemaInfo(kvSchemaInfo);
 
         assertEquals(schemaInfoKeyValue.getKey(), FOO_SCHEMA.getSchemaInfo());
         assertEquals(schemaInfoKeyValue.getValue(), BAR_SCHEMA.getSchemaInfo());
@@ -138,26 +139,26 @@ public void encodeDecodeNestedKeyValueSchemaInfo(KeyValueEncodingType encodingTy
         );
         SchemaInfo kvSchemaInfo = kvSchema.getSchemaInfo();
         assertEquals(
-            DefaultImplementation.decodeKeyValueEncodingType(kvSchemaInfo),
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueEncodingType(kvSchemaInfo),
             encodingType);
 
         SchemaInfo encodedSchemaInfo =
-            DefaultImplementation.encodeKeyValueSchemaInfo(
+                DefaultImplementation.getDefaultImplementation().encodeKeyValueSchemaInfo(
                 FOO_SCHEMA,
                 nestedSchema,
                 encodingType);
         assertEquals(encodedSchemaInfo, kvSchemaInfo);
         assertEquals(
-            DefaultImplementation.decodeKeyValueEncodingType(encodedSchemaInfo),
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueEncodingType(encodedSchemaInfo),
             encodingType);
 
         KeyValue<SchemaInfo, SchemaInfo> schemaInfoKeyValue =
-            DefaultImplementation.decodeKeyValueSchemaInfo(kvSchemaInfo);
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueSchemaInfo(kvSchemaInfo);
 
         assertEquals(schemaInfoKeyValue.getKey(), FOO_SCHEMA.getSchemaInfo());
         assertEquals(schemaInfoKeyValue.getValue().getType(), SchemaType.KEY_VALUE);
         KeyValue<SchemaInfo, SchemaInfo> nestedSchemaInfoKeyValue =
-            DefaultImplementation.decodeKeyValueSchemaInfo(schemaInfoKeyValue.getValue());
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueSchemaInfo(schemaInfoKeyValue.getValue());
 
         assertEquals(nestedSchemaInfoKeyValue.getKey(), Schema.STRING.getSchemaInfo());
         assertEquals(nestedSchemaInfoKeyValue.getValue(), BAR_SCHEMA.getSchemaInfo());
@@ -178,11 +179,11 @@ public void testKeyValueSchemaInfoBackwardCompatibility() {
             .setProperties(Collections.emptyMap());
 
         assertEquals(
-            DefaultImplementation.decodeKeyValueEncodingType(oldSchemaInfo),
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueEncodingType(oldSchemaInfo),
             KeyValueEncodingType.INLINE);
 
         KeyValue<SchemaInfo, SchemaInfo> schemaInfoKeyValue =
-            DefaultImplementation.decodeKeyValueSchemaInfo(oldSchemaInfo);
+                DefaultImplementation.getDefaultImplementation().decodeKeyValueSchemaInfo(oldSchemaInfo);
         // verify the key schema
         SchemaInfo keySchemaInfo = schemaInfoKeyValue.getKey();
         assertEquals(
@@ -210,14 +211,14 @@ public void testKeyValueSchemaInfoBackwardCompatibility() {
     }
 
     @Test
-    public void testKeyValueSchemaInfoToString() throws JSONException {
-        String havePrimitiveType = DefaultImplementation
+    public void testKeyValueSchemaInfoToString() throws Exception {
+        String havePrimitiveType = DefaultImplementation.getDefaultImplementation()
                 .convertKeyValueSchemaInfoDataToString(KeyValueSchemaInfo
                         .decodeKeyValueSchemaInfo(Schema.KeyValue(Schema.AVRO(Foo.class), Schema.STRING)
                                 .getSchemaInfo()));
         JSONSchemaTest.assertJSONEqual(havePrimitiveType, KEY_VALUE_SCHEMA_INFO_INCLUDE_PRIMITIVE);
 
-        String notHavePrimitiveType = DefaultImplementation
+        String notHavePrimitiveType = DefaultImplementation.getDefaultImplementation()
                 .convertKeyValueSchemaInfoDataToString(KeyValueSchemaInfo
                         .decodeKeyValueSchemaInfo(Schema.KeyValue(Schema.AVRO(Foo.class),
                                 Schema.AVRO(Foo.class)).getSchemaInfo()));
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaUtils.java b/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaUtils.java
index 23ecb4eb4691b..9843e638a7d62 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaUtils.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaUtils.java
@@ -287,7 +287,7 @@ public JsonElement serialize(byte[] src, Type typeOfSrc, JsonSerializationContex
                     return toJsonObject(schemaInfo.getSchemaDefinition());
                 case KEY_VALUE:
                     KeyValue<SchemaInfo, SchemaInfo> schemaInfoKeyValue =
-                        DefaultImplementation.decodeKeyValueSchemaInfo(schemaInfo);
+                        DefaultImplementation.getDefaultImplementation().decodeKeyValueSchemaInfo(schemaInfo);
                     JsonObject obj = new JsonObject();
                     String keyJson = jsonifySchemaInfo(schemaInfoKeyValue.getKey());
                     String valueJson = jsonifySchemaInfo(schemaInfoKeyValue.getValue());