diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
index 6863a34..ae1bda5 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
@@ -436,7 +436,7 @@
     }
 
     public SchemaHash getSchemaHash() {
-        return schemaHash == null ? SchemaHash.of(new byte[0], null) : schemaHash;
+        return schemaHash == null ? SchemaHash.empty() : schemaHash;
     }
 
     public void setSchemaInfoForReplicator(SchemaInfo schemaInfo) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/BooleanSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/BooleanSchema.java
index 0c41e1f..49548c1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/BooleanSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/BooleanSchema.java
@@ -32,10 +32,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-                .setName("Boolean")
-                .setType(SchemaType.BOOLEAN)
-                .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+                .name("Boolean")
+                .type(SchemaType.BOOLEAN)
+                .schema(new byte[0]).build();
         INSTANCE = new BooleanSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java
index 7665d96..73431ae 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java
@@ -33,10 +33,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-            .setName("ByteBuf")
-            .setType(SchemaType.BYTES)
-            .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+            .name("ByteBuf")
+            .type(SchemaType.BYTES)
+            .schema(new byte[0]).build();
         INSTANCE = new ByteBufSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufferSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufferSchema.java
index 1518413..0bc4e9c 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufferSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufferSchema.java
@@ -32,10 +32,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-            .setName("ByteBuffer")
-            .setType(SchemaType.BYTES)
-            .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+            .name("ByteBuffer")
+            .type(SchemaType.BYTES)
+            .schema(new byte[0]).build();
         INSTANCE = new ByteBufferSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteSchema.java
index 6d51687..3e56381 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteSchema.java
@@ -32,10 +32,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-            .setName("INT8")
-            .setType(SchemaType.INT8)
-            .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+            .name("INT8")
+            .type(SchemaType.INT8)
+            .schema(new byte[0]).build();
         INSTANCE = new ByteSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/BytesSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/BytesSchema.java
index 98a0e66..5706af9 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/BytesSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/BytesSchema.java
@@ -31,10 +31,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-            .setName("Bytes")
-            .setType(SchemaType.BYTES)
-            .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+            .name("Bytes")
+            .type(SchemaType.BYTES)
+            .schema(new byte[0]).build();
         INSTANCE = new BytesSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/DateSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/DateSchema.java
index f632b99..7997691 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/DateSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/DateSchema.java
@@ -32,10 +32,10 @@
    private static final SchemaInfo SCHEMA_INFO;
 
    static {
-       SCHEMA_INFO = new SchemaInfoImpl()
-             .setName("Date")
-             .setType(SchemaType.DATE)
-             .setSchema(new byte[0]);
+       SCHEMA_INFO = SchemaInfoImpl.builder()
+             .name("Date")
+             .type(SchemaType.DATE)
+             .schema(new byte[0]).build();
        INSTANCE = new DateSchema();
    }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/DoubleSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/DoubleSchema.java
index d38deb4..b5d8076 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/DoubleSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/DoubleSchema.java
@@ -32,10 +32,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-            .setName("Double")
-            .setType(SchemaType.DOUBLE)
-            .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+            .name("Double")
+            .type(SchemaType.DOUBLE)
+            .schema(new byte[0]).build();
         INSTANCE = new DoubleSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/FloatSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/FloatSchema.java
index 84d4073..0d26411 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/FloatSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/FloatSchema.java
@@ -32,10 +32,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-                .setName("Float")
-                .setType(SchemaType.FLOAT)
-                .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+                .name("Float")
+                .type(SchemaType.FLOAT)
+                .schema(new byte[0]).build();
         INSTANCE = new FloatSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/InstantSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/InstantSchema.java
index 8adf7c1..569a938 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/InstantSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/InstantSchema.java
@@ -33,10 +33,10 @@
    private static final SchemaInfo SCHEMA_INFO;
 
    static {
-       SCHEMA_INFO = new SchemaInfoImpl()
-             .setName("Instant")
-             .setType(SchemaType.INSTANT)
-             .setSchema(new byte[0]);
+       SCHEMA_INFO = SchemaInfoImpl.builder()
+             .name("Instant")
+             .type(SchemaType.INSTANT)
+             .schema(new byte[0]).build();
        INSTANCE = new InstantSchema();
    }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/IntSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/IntSchema.java
index dfad280..5212565 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/IntSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/IntSchema.java
@@ -32,10 +32,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-            .setName("INT32")
-            .setType(SchemaType.INT32)
-            .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+            .name("INT32")
+            .type(SchemaType.INT32)
+            .schema(new byte[0]).build();
         INSTANCE = new IntSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java
index 4f09ffe..ad421d8 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java
@@ -71,11 +71,12 @@
             ObjectMapper objectMapper = new ObjectMapper();
             JsonSchemaGenerator schemaGen = new JsonSchemaGenerator(objectMapper);
             JsonSchema jsonBackwardsCompatibleSchema = schemaGen.generateSchema(pojo);
-            backwardsCompatibleSchemaInfo = new SchemaInfoImpl()
-                    .setName("")
-                    .setProperties(schemaInfo.getProperties())
-                    .setType(SchemaType.JSON)
-                    .setSchema(objectMapper.writeValueAsBytes(jsonBackwardsCompatibleSchema));
+            backwardsCompatibleSchemaInfo = SchemaInfoImpl.builder()
+                    .name("")
+                    .properties(schemaInfo.getProperties())
+                    .type(SchemaType.JSON)
+                    .schema(objectMapper.writeValueAsBytes(jsonBackwardsCompatibleSchema))
+                    .build();
         } catch (JsonProcessingException ex) {
             throw new RuntimeException(ex);
         }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalDateSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalDateSchema.java
index 4115eee..5c0420a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalDateSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalDateSchema.java
@@ -32,10 +32,10 @@
    private static final SchemaInfo SCHEMA_INFO;
 
    static {
-       SCHEMA_INFO = new SchemaInfoImpl()
-             .setName("LocalDate")
-             .setType(SchemaType.LOCAL_DATE)
-             .setSchema(new byte[0]);
+       SCHEMA_INFO = SchemaInfoImpl.builder()
+             .name("LocalDate")
+             .type(SchemaType.LOCAL_DATE)
+             .schema(new byte[0]).build();
        INSTANCE = new LocalDateSchema();
    }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalDateTimeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalDateTimeSchema.java
index a6bac5f..e7c5965 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalDateTimeSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalDateTimeSchema.java
@@ -36,10 +36,10 @@
    public static final String DELIMITER = ":";
 
    static {
-       SCHEMA_INFO = new SchemaInfoImpl()
-             .setName("LocalDateTime")
-             .setType(SchemaType.LOCAL_DATE_TIME)
-             .setSchema(new byte[0]);
+       SCHEMA_INFO = SchemaInfoImpl.builder()
+             .name("LocalDateTime")
+             .type(SchemaType.LOCAL_DATE_TIME)
+             .schema(new byte[0]).build();
        INSTANCE = new LocalDateTimeSchema();
    }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalTimeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalTimeSchema.java
index cdf0663..59ba5d2 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalTimeSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LocalTimeSchema.java
@@ -32,10 +32,10 @@
    private static final SchemaInfo SCHEMA_INFO;
 
    static {
-       SCHEMA_INFO = new SchemaInfoImpl()
-             .setName("LocalTime")
-             .setType(SchemaType.LOCAL_TIME)
-             .setSchema(new byte[0]);
+       SCHEMA_INFO = SchemaInfoImpl.builder()
+             .name("LocalTime")
+             .type(SchemaType.LOCAL_TIME)
+             .schema(new byte[0]).build();
        INSTANCE = new LocalTimeSchema();
    }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LongSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LongSchema.java
index deccaf4..e136fb3 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LongSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/LongSchema.java
@@ -32,10 +32,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-            .setName("INT64")
-            .setType(SchemaType.INT64)
-            .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+            .name("INT64")
+            .type(SchemaType.INT64)
+            .schema(new byte[0]).build();
         INSTANCE = new LongSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ShortSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ShortSchema.java
index bbb5ad6..afa0f9a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ShortSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ShortSchema.java
@@ -32,10 +32,10 @@
     private static final SchemaInfo SCHEMA_INFO;
 
     static {
-        SCHEMA_INFO = new SchemaInfoImpl()
-            .setName("INT16")
-            .setType(SchemaType.INT16)
-            .setSchema(new byte[0]);
+        SCHEMA_INFO = SchemaInfoImpl.builder()
+            .name("INT16")
+            .type(SchemaType.INT16)
+            .schema(new byte[0]).build();
         INSTANCE = new ShortSchema();
     }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StringSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StringSchema.java
index 370946a..2096f83 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StringSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StringSchema.java
@@ -43,10 +43,10 @@
         // Ensure the ordering of the static initialization
         CHARSET_KEY = "__charset";
         DEFAULT_CHARSET = StandardCharsets.UTF_8;
-        DEFAULT_SCHEMA_INFO = new SchemaInfoImpl()
-                .setName("String")
-                .setType(SchemaType.STRING)
-                .setSchema(new byte[0]);
+        DEFAULT_SCHEMA_INFO = SchemaInfoImpl.builder()
+                .name("String")
+                .type(SchemaType.STRING)
+                .schema(new byte[0]).build();
 
         UTF8 = new StringSchema(StandardCharsets.UTF_8);
     }
@@ -84,11 +84,12 @@
         this.charset = charset;
         Map<String, String> properties = new HashMap<>();
         properties.put(CHARSET_KEY, charset.name());
-        this.schemaInfo = new SchemaInfoImpl()
-                .setName(DEFAULT_SCHEMA_INFO.getName())
-                .setType(SchemaType.STRING)
-                .setSchema(DEFAULT_SCHEMA_INFO.getSchema())
-                .setProperties(properties);
+        this.schemaInfo = SchemaInfoImpl.builder()
+                .name(DEFAULT_SCHEMA_INFO.getName())
+                .type(SchemaType.STRING)
+                .schema(DEFAULT_SCHEMA_INFO.getSchema())
+                .properties(properties)
+                .build();
     }
 
     public byte[] encode(String message) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/TimeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/TimeSchema.java
index 2be6e9d..43fc069 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/TimeSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/TimeSchema.java
@@ -32,10 +32,10 @@
    private static final SchemaInfo SCHEMA_INFO;
 
    static {
-       SCHEMA_INFO = new SchemaInfoImpl()
-             .setName("Time")
-             .setType(SchemaType.TIME)
-             .setSchema(new byte[0]);
+       SCHEMA_INFO = SchemaInfoImpl.builder()
+             .name("Time")
+             .type(SchemaType.TIME)
+             .schema(new byte[0]).build();
        INSTANCE = new TimeSchema();
    }
 
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/TimestampSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/TimestampSchema.java
index 0b42d47..c89dfef 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/TimestampSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/TimestampSchema.java
@@ -32,10 +32,10 @@
    private static final SchemaInfo SCHEMA_INFO;
 
    static {
-       SCHEMA_INFO = new SchemaInfoImpl()
-             .setName("Timestamp")
-             .setType(SchemaType.TIMESTAMP)
-             .setSchema(new byte[0]);
+       SCHEMA_INFO = SchemaInfoImpl.builder()
+             .name("Timestamp")
+             .type(SchemaType.TIMESTAMP)
+             .schema(new byte[0]).build();
        INSTANCE = new TimestampSchema();
    }
 
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 188c1e7..1691569 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
@@ -170,11 +170,11 @@
             KeyValueEncodingType.SEPARATED
         );
 
-        SchemaInfo oldSchemaInfo = new SchemaInfoImpl()
-            .setName("")
-            .setType(SchemaType.KEY_VALUE)
-            .setSchema(kvSchema.getSchemaInfo().getSchema())
-            .setProperties(Collections.emptyMap());
+        SchemaInfo oldSchemaInfo = SchemaInfoImpl.builder()
+            .name("")
+            .type(SchemaType.KEY_VALUE)
+            .schema(kvSchema.getSchemaInfo().getSchema())
+            .properties(Collections.emptyMap()).build();
 
         assertEquals(
                 DefaultImplementation.getDefaultImplementation().decodeKeyValueEncodingType(oldSchemaInfo),
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/StringSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/StringSchemaTest.java
index 8a98974..4b22085 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/StringSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/StringSchemaTest.java
@@ -86,11 +86,11 @@
 
     @Test
     public void testSchemaInfoWithoutCharset() {
-        SchemaInfo si = new SchemaInfoImpl()
-            .setName("test-schema-info-without-charset")
-            .setType(SchemaType.STRING)
-            .setSchema(new byte[0])
-            .setProperties(Collections.emptyMap());
+        SchemaInfo si = SchemaInfoImpl.builder()
+            .name("test-schema-info-without-charset")
+            .type(SchemaType.STRING)
+            .schema(new byte[0])
+            .properties(Collections.emptyMap()).build();
         StringSchema schema = StringSchema.fromSchemaInfo(si);
 
         String myString = "my string for test";
@@ -121,11 +121,11 @@
     public void testSchemaInfoWithCharset(Charset charset) {
         Map<String, String> properties = new HashMap<>();
         properties.put(StringSchema.CHARSET_KEY, charset.name());
-        SchemaInfo si = new SchemaInfoImpl()
-            .setName("test-schema-info-without-charset")
-            .setType(SchemaType.STRING)
-            .setSchema(new byte[0])
-            .setProperties(properties);
+        SchemaInfo si = SchemaInfoImpl.builder()
+            .name("test-schema-info-without-charset")
+            .type(SchemaType.STRING)
+            .schema(new byte[0])
+            .properties(properties).build();
         StringSchema schema = StringSchema.fromSchemaInfo(si);
 
         String myString = "my string for test";
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaInfoImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaInfoImpl.java
index d67dc5f..b06ee51 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaInfoImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/client/impl/schema/SchemaInfoImpl.java
@@ -19,10 +19,10 @@
 package org.apache.pulsar.client.impl.schema;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.Map;
-import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -30,6 +30,7 @@
 import lombok.experimental.Accessors;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
+import org.apache.pulsar.common.protocol.schema.SchemaHash;
 import org.apache.pulsar.common.schema.KeyValue;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
@@ -40,10 +41,8 @@
 @InterfaceAudience.Public
 @InterfaceStability.Stable
 @Data
-@AllArgsConstructor
 @NoArgsConstructor
 @Accessors(chain = true)
-@Builder
 public class SchemaInfoImpl implements SchemaInfo {
 
     @EqualsAndHashCode.Exclude
@@ -67,9 +66,23 @@
     /**
      * Additional properties of the schema definition (implementation defined).
      */
-    @Builder.Default
     private Map<String, String> properties = Collections.emptyMap();
 
+    @EqualsAndHashCode.Exclude
+    @JsonIgnore
+    private transient SchemaHash schemaHash;
+
+    @Builder
+    public SchemaInfoImpl(String name, byte[] schema, SchemaType type, long timestamp,
+                          Map<String, String> properties) {
+        this.name = name;
+        this.schema = schema;
+        this.type = type;
+        this.timestamp = timestamp;
+        this.properties = properties == null ? Collections.emptyMap() : properties;
+        this.schemaHash = SchemaHash.of(this.schema, this.type);
+    }
+
     public String getSchemaDefinition() {
         if (null == schema) {
             return "";
@@ -89,6 +102,19 @@
         }
     }
 
+    /**
+     * Calculate the SchemaHash for compatible with `@NoArgsConstructor`.
+     * If SchemaInfoImpl is created by no-args-constructor from users, the schemaHash will be null.
+     * Note: We should remove this method as long as `@NoArgsConstructor` removed at major release to avoid null-check
+     * overhead.
+     */
+    public SchemaHash getSchemaHash() {
+        if (schemaHash == null) {
+            schemaHash = SchemaHash.of(this.schema, this.type);
+        }
+        return schemaHash;
+    }
+
     @Override
     public String toString() {
         return SchemaUtils.jsonifySchemaInfo(this);
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/SchemaHash.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/SchemaHash.java
index 8bbc18f..46f92b1 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/SchemaHash.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/SchemaHash.java
@@ -21,9 +21,9 @@
 import com.google.common.hash.HashCode;
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
-import java.util.Optional;
 import lombok.EqualsAndHashCode;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.schema.SchemaInfoImpl;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
 
@@ -33,7 +33,8 @@
 @EqualsAndHashCode
 public class SchemaHash {
 
-    private static HashFunction hashFunction = Hashing.sha256();
+    private static final HashFunction hashFunction = Hashing.sha256();
+    private static final SchemaHash EMPTY_SCHEMA_HASH = new SchemaHash(hashFunction.hashBytes(new byte[0]), null);
 
     private final HashCode hash;
 
@@ -45,9 +46,10 @@
     }
 
     public static SchemaHash of(Schema schema) {
-        Optional<SchemaInfo> schemaInfo = Optional.ofNullable(schema).map(Schema::getSchemaInfo);
-        return of(schemaInfo.map(SchemaInfo::getSchema).orElseGet(() -> new byte[0]),
-                schemaInfo.map(SchemaInfo::getType).orElse(null));
+        if (schema == null || schema.getSchemaInfo() == null) {
+            return EMPTY_SCHEMA_HASH;
+        }
+        return ((SchemaInfoImpl) schema.getSchemaInfo()).getSchemaHash();
     }
 
     public static SchemaHash of(SchemaData schemaData) {
@@ -55,12 +57,19 @@
     }
 
     public static SchemaHash of(SchemaInfo schemaInfo) {
-        return of(schemaInfo == null ? new byte[0] : schemaInfo.getSchema(),
-                schemaInfo == null ? null : schemaInfo.getType());
+        if (schemaInfo == null) {
+            return EMPTY_SCHEMA_HASH;
+        }
+        return ((SchemaInfoImpl) schemaInfo).getSchemaHash();
     }
 
+    public static SchemaHash empty() {
+        return EMPTY_SCHEMA_HASH;
+    }
+
+    // Shouldn't call this method frequently, otherwise will bring performance regression
     public static SchemaHash of(byte[] schemaBytes, SchemaType schemaType) {
-        return new SchemaHash(hashFunction.hashBytes(schemaBytes), schemaType);
+        return new SchemaHash(hashFunction.hashBytes(schemaBytes == null ? new byte[0] : schemaBytes), schemaType);
     }
 
     public byte[] asBytes() {