add code.

Author: thetumbled

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

@@ -2801,13 +2801,24 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
     )
     private long brokerServiceCompactionThresholdInBytes = 0;
 
+    @Deprecated
     @FieldContext(
             category = CATEGORY_SERVER,
+            deprecated = true,
             doc = "Timeout for the compaction phase one loop, If the execution time of the compaction "
                     + "phase one loop exceeds this time, the compaction will not proceed."
+                    + "@deprecated - Use brokerServiceCompactionPhaseOneLoopReadTimeoutInSeconds instead" +
+                    "to control the timeout for each read request."
     )
     private long brokerServiceCompactionPhaseOneLoopTimeInSeconds = 30;
 
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Timeout for each read request in the compaction phase one loop, If the execution time of one "
+                    + "single message read operation exceeds this time, the compaction will not proceed."
+    )
+    private long brokerServiceCompactionPhaseOneLoopReadTimeoutInSeconds = 30;
+
     @FieldContext(
             category = CATEGORY_SERVER,
             doc = "Whether retain null-key message during topic compaction."

pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java

@@ -72,7 +72,7 @@ public StrategicTwoPhaseCompactor(ServiceConfiguration conf,
                                       ScheduledExecutorService scheduler) {
         super(conf, pulsar, bk, scheduler);
         batchMessageContainer = new RawBatchMessageContainerImpl();
-        phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+        phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopReadTimeoutInSeconds());
     }
 
     public CompletableFuture<Long> compact(String topic) {

pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java

@@ -70,7 +70,7 @@ public TwoPhaseCompactor(ServiceConfiguration conf,
                              BookKeeper bk,
                              ScheduledExecutorService scheduler) {
         super(conf, pulsar, bk, scheduler);
-        phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopTimeInSeconds());
+        phaseOneLoopReadTimeout = Duration.ofSeconds(conf.getBrokerServiceCompactionPhaseOneLoopReadTimeoutInSeconds());
         topicCompactionRetainNullKey = conf.isTopicCompactionRetainNullKey();
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java

@@ -260,7 +260,7 @@ public void testCompactEmptyTopic() throws Exception {
     @Test
     public void testPhaseOneLoopTimeConfiguration() {
         ServiceConfiguration configuration = new ServiceConfiguration();
-        configuration.setBrokerServiceCompactionPhaseOneLoopTimeInSeconds(60);
+        configuration.setBrokerServiceCompactionPhaseOneLoopReadTimeoutInSeconds(60);
         PulsarClientImpl mockClient = mock(PulsarClientImpl.class);
         ConnectionPool connectionPool = mock(ConnectionPool.class);
         when(mockClient.getCnxPool()).thenReturn(connectionPool);