diff --git a/conf/broker.conf b/conf/broker.conf
index 0efc6569abfb5..47795a56bca30 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -487,6 +487,10 @@ maxMessagePublishBufferSizeInMB=
 # Use 0 or negative number to disable the check
 retentionCheckIntervalInSeconds=120
 
+# Control the frequency of checking the max message size in the topic policy. 
+# The default interval is 60 seconds.
+maxMessageSizeCheckIntervalInSeconds=60
+
 # Max number of partitions per partitioned topic
 # Use 0 or negative number to disable the check
 maxNumPartitionsPerPartitionedTopic=0
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 16a877dc62899..2ef53d6169aee 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -118,6 +118,10 @@ maxPendingPublishRequestsPerConnection=1000
 # How frequently to proactively check and purge expired messages
 messageExpiryCheckIntervalInMinutes=5
 
+# Check between intervals to see if max message size in topic policies has been updated.
+# Default is 60s
+maxMessageSizeCheckIntervalInSeconds=60
+
 # How long to delay rewinding cursor and dispatching messages when active consumer is changed
 activeConsumerFailoverDelayTimeMillis=1000
 
diff --git a/deployment/terraform-ansible/templates/broker.conf b/deployment/terraform-ansible/templates/broker.conf
index b17684dc38f2e..1e0e43fc943c1 100644
--- a/deployment/terraform-ansible/templates/broker.conf
+++ b/deployment/terraform-ansible/templates/broker.conf
@@ -412,6 +412,10 @@ messagePublishBufferCheckIntervalInMillis=100
 # Use 0 or negative number to disable the check
 retentionCheckIntervalInSeconds=120
 
+# Check between intervals to see if max message size in topic policies has been updated.
+# Default is 60s
+maxMessageSizeCheckIntervalInSeconds=60
+
 # Max number of partitions per partitioned topic
 # Use 0 or negative number to disable the check
 maxNumPartitionsPerPartitionedTopic=0
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index b41f76fc6a0d0..9b5f9399e16c8 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -916,6 +916,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private int retentionCheckIntervalInSeconds = 120;
 
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Check between intervals to see if max message size of topic policy has updated. default is 60s"
+    )
+    private int maxMessageSizeCheckIntervalInSeconds = 60;
+
     @FieldContext(
             category = CATEGORY_SERVER,
             doc = "The number of partitions per partitioned topic.\n"
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index e2133f6345fb4..33c973cbcc1e5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -120,6 +120,10 @@ public abstract class AbstractTopic implements Topic {
             AtomicLongFieldUpdater.newUpdater(AbstractTopic.class, "usageCount");
     private volatile long usageCount = 0;
 
+    private volatile int topicMaxMessageSize = 0;
+    private volatile long lastTopicMaxMessageSizeCheckTimeStamp = 0;
+    private final long topicMaxMessageSizeCheckIntervalMs;
+
     public AbstractTopic(String topic, BrokerService brokerService) {
         this.topic = topic;
         this.brokerService = brokerService;
@@ -132,6 +136,8 @@ public AbstractTopic(String topic, BrokerService brokerService) {
                 .getBrokerDeleteInactiveTopicsMaxInactiveDurationSeconds());
         this.inactiveTopicPolicies.setInactiveTopicDeleteMode(brokerService.pulsar().getConfiguration()
                 .getBrokerDeleteInactiveTopicsMode());
+        this.topicMaxMessageSizeCheckIntervalMs = TimeUnit.SECONDS.toMillis(brokerService.pulsar().getConfiguration()
+                .getMaxMessageSizeCheckIntervalInSeconds());
         this.lastActive = System.nanoTime();
         Policies policies = null;
         try {
@@ -853,14 +859,16 @@ protected int getWaitingProducersCount() {
     }
 
     protected boolean isExceedMaximumMessageSize(int size) {
-        return getTopicPolicies()
-                .map(TopicPolicies::getMaxMessageSize)
-                .map(maxMessageSize -> {
-                    if (maxMessageSize == 0) {
-                        return false;
-                    }
-                    return size > maxMessageSize;
-                }).orElse(false);
+        if (lastTopicMaxMessageSizeCheckTimeStamp + topicMaxMessageSizeCheckIntervalMs < System.currentTimeMillis()) {
+            // refresh topicMaxMessageSize from topic policies
+            topicMaxMessageSize = getTopicPolicies().map(TopicPolicies::getMaxMessageSize).orElse(0);
+            lastTopicMaxMessageSizeCheckTimeStamp = System.currentTimeMillis();
+        }
+
+        if (topicMaxMessageSize == 0) {
+            return false;
+        }
+        return size > topicMaxMessageSize;
     }
 
     /**
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index 3934d3b342c4e..fd664aeb164ae 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -100,6 +100,7 @@ protected void setup() throws Exception {
         this.conf.setSystemTopicEnabled(true);
         this.conf.setTopicLevelPoliciesEnabled(true);
         this.conf.setDefaultNumberOfNamespaceBundles(1);
+        this.conf.setMaxMessageSizeCheckIntervalInSeconds(1);
         super.internalSetup();
 
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
@@ -1781,8 +1782,14 @@ private void doTestTopicMaxMessageSize(boolean isPartitioned) throws Exception {
             assertEquals(e.getStatusCode(), 412);
         }
 
-        MessageId messageId = producer.send(new byte[1024]);
-        assertNotNull(messageId);
+        Awaitility.await().untilAsserted(() -> {
+            try {
+                MessageId messageId = producer.send(new byte[1024]);
+                assertNotNull(messageId);
+            } catch (PulsarClientException e) {
+                fail("failed to send message");
+            }
+        });
         producer.close();
     }