Add opt-in drainProducerRate to throttle producers during the backlog drain

benchmark-framework/src/main/java/io/openmessaging/benchmark/Workload.java

@@ -67,6 +67,14 @@ public class Workload {
      */
     public long consumerBacklogSizeGB = 0;
 
+    /**
+     * Producer publish rate (msg/s) to apply during the backlog-drain phase.
+     * Default -1 means "unset" — producers keep producerRate through the drain
+     * (original behavior). >= 0 throttles producers for the drain only; 0 is
+     * floored to ~1 msg/s by Worker.adjustPublishRate (effectively stopped).
+     */
+    public int drainProducerRate = -1;
+
     public int warmupDurationMinutes = 30;
     public int sampleRateMillis = 10000;
     public int testDurationMinutes;

benchmark-framework/src/main/java/io/openmessaging/benchmark/WorkloadGenerator.java

@@ -391,6 +391,11 @@ private void buildAndDrainBacklog(List<String> topics) throws IOException {
 
         log.info("--- Start draining backlog ---");
 
+        if (workload.drainProducerRate >= 0) {
+            log.info("Throttling producers to {} msg/s for backlog drain", workload.drainProducerRate);
+            worker.adjustPublishRate(workload.drainProducerRate);
+        }
+
         worker.resumeConsumers();
 
         final long minBacklog = 1000;

benchmark-framework/src/test/java/io/openmessaging/benchmark/WorkloadTest.java

@@ -0,0 +1,33 @@
+package io.openmessaging.benchmark;
+
+import static org.junit.Assert.assertEquals;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import org.junit.Test;
+
+public class WorkloadTest {
+
+    // Mirror the strict YAML workload mapper used in Benchmark.java.
+    private static final ObjectMapper MAPPER = new ObjectMapper(new YAMLFactory());
+
+    private static final String BASE =
+            "name: drain-test\n"
+          + "topics: 1\n"
+          + "partitionsPerTopic: 1\n"
+          + "messageSize: 100\n"
+          + "producerRate: 1000\n"
+          + "testDurationMinutes: 1\n";
+
+    @Test
+    public void drainProducerRateDefaultsToUnset() throws Exception {
+        Workload w = MAPPER.readValue(BASE, Workload.class);
+        assertEquals(-1, w.drainProducerRate);
+    }
+
+    @Test
+    public void drainProducerRateParsesWhenPresent() throws Exception {
+        Workload w = MAPPER.readValue(BASE + "drainProducerRate: 5000\n", Workload.class);
+        assertEquals(5000, w.drainProducerRate);
+    }
+}