[improve] [broker] Enable AppendBrokerTimestampMetadataInterceptor by default

conf/broker.conf

@@ -1843,5 +1843,10 @@ brokerInterceptorsDirectory=./interceptors
 # List of broker interceptor to load, which is a list of broker interceptor names
 brokerInterceptors=
 
-# Enable or disable the broker interceptor, which is only used for testing for now
-disableBrokerInterceptors=true
+# Enable or disable the broker interceptor.
+disableBrokerInterceptors=true
+
+# List of interceptors for entry metadata
+# AppendBrokerTimestampMetadataInterceptor is enabled by default and it can make message TTL more accurate,
+# but it increase 8 bytes to each message.
+brokerEntryMetadataInterceptors=org.apache.pulsar.common.intercept.AppendBrokerTimestampMetadataInterceptor

deployment/terraform-ansible/templates/broker.conf

@@ -1154,3 +1154,8 @@ managedLedgerCacheEvictionFrequency=0
 # Enable transaction coordinator in broker
 transactionCoordinatorEnabled=true
 transactionMetadataStoreProviderClassName=org.apache.pulsar.transaction.coordinator.impl.InMemTransactionMetadataStoreProvider
+
+# List of interceptors for entry metadata
+# AppendBrokerTimestampMetadataInterceptor is enabled by default and it can make message TTL more accurate,
+# but it increase 8 bytes to each message.
+brokerEntryMetadataInterceptors=org.apache.pulsar.common.intercept.AppendBrokerTimestampMetadataInterceptor

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -21,7 +21,6 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Objects;
@@ -1439,8 +1438,11 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
 
     @FieldContext(
             category = CATEGORY_SERVER,
-            doc = "List of interceptors for entry metadata.")
-    private Set<String> brokerEntryMetadataInterceptors = new HashSet<>();
+            doc = "List of interceptors for entry metadata, "
+                    + "AppendBrokerTimestampMetadataInterceptor is enabled by default and it can make message "
+                    + "TTL more accurate, but it increase 8 bytes to each message.")
+    private Set<String> brokerEntryMetadataInterceptors =
+            Sets.newHashSet("org.apache.pulsar.common.intercept.AppendBrokerTimestampMetadataInterceptor");
 
     @FieldContext(
             category = CATEGORY_SERVER,

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerInternalClientConfigurationOverrideTest.java

@@ -18,7 +18,9 @@
  */
 package org.apache.pulsar.broker.service;
 
+import static org.testng.Assert.assertEquals;
 import org.apache.pulsar.broker.PulsarServerException;
+import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.client.admin.internal.PulsarAdminImpl;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
@@ -112,4 +114,12 @@ public void testNamespaceServicePulsarClientConfiguration() {
         Assert.assertEquals(clientConf.getMemoryLimitBytes(), 100000);
     }
 
+    @Test
+    public void testEnableInterceptor() {
+        assertEquals(conf.getBrokerEntryMetadataInterceptors().iterator().next(),
+                "org.apache.pulsar.common.intercept.AppendBrokerTimestampMetadataInterceptor");
+        ServiceConfiguration configuration = new ServiceConfiguration();
+        assertEquals(configuration.getBrokerEntryMetadataInterceptors().iterator().next(),
+                "org.apache.pulsar.common.intercept.AppendBrokerTimestampMetadataInterceptor");
+    }
 }