From 25936fc6bf71da2da5333674bdf11baecaf1662a Mon Sep 17 00:00:00 2001
From: maxliu <ly_chinese@163.com>
Date: Wed, 10 Apr 2024 13:08:20 +0800
Subject: [PATCH] [segment replication] decouple the rateLimiter of segrep and
 recovery (#12939)

add setting "indices.replication.max_bytes_per_sec" which takes effect when not negative

Signed-off-by: maxliu <ly_chinese@163.com>
---
 .../replication/RemoteSegmentFileChunkWriter.java      | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/server/src/main/java/org/opensearch/indices/replication/RemoteSegmentFileChunkWriter.java b/server/src/main/java/org/opensearch/indices/replication/RemoteSegmentFileChunkWriter.java
index 896fb77ddac1a..179e497565326 100644
--- a/server/src/main/java/org/opensearch/indices/replication/RemoteSegmentFileChunkWriter.java
+++ b/server/src/main/java/org/opensearch/indices/replication/RemoteSegmentFileChunkWriter.java
@@ -37,12 +37,11 @@ public final class RemoteSegmentFileChunkWriter implements FileChunkWriter {
     private final AtomicLong requestSeqNoGenerator;
     private final RetryableTransportClient retryableTransportClient;
     private final ShardId shardId;
-    private final RecoverySettings recoverySettings;
     private final long replicationId;
     private final AtomicLong bytesSinceLastPause = new AtomicLong();
     private final TransportRequestOptions fileChunkRequestOptions;
     private final Consumer<Long> onSourceThrottle;
-    private final Supplier<RateLimiter> rateLimiterProvider;
+    private final Supplier<RateLimiter> rateLimiterSupplier;
     private final String action;
 
     public RemoteSegmentFileChunkWriter(
@@ -53,15 +52,14 @@ public RemoteSegmentFileChunkWriter(
         String action,
         AtomicLong requestSeqNoGenerator,
         Consumer<Long> onSourceThrottle,
-        Supplier<RateLimiter> rateLimiterProvider
+        Supplier<RateLimiter> rateLimiterSupplier
     ) {
         this.replicationId = replicationId;
-        this.recoverySettings = recoverySettings;
         this.retryableTransportClient = retryableTransportClient;
         this.shardId = shardId;
         this.requestSeqNoGenerator = requestSeqNoGenerator;
         this.onSourceThrottle = onSourceThrottle;
-        this.rateLimiterProvider = rateLimiterProvider;
+        this.rateLimiterSupplier = rateLimiterSupplier;
         this.fileChunkRequestOptions = TransportRequestOptions.builder()
             .withType(TransportRequestOptions.Type.RECOVERY)
             .withTimeout(recoverySettings.internalActionTimeout())
@@ -82,7 +80,7 @@ public void writeFileChunk(
         // Pause using the rate limiter, if desired, to throttle the recovery
         final long throttleTimeInNanos;
         // always fetch the ratelimiter - it might be updated in real-time on the recovery settings
-        final RateLimiter rl = rateLimiterProvider.get();
+        final RateLimiter rl = rateLimiterSupplier.get();
         if (rl != null) {
             long bytes = bytesSinceLastPause.addAndGet(content.length());
             if (bytes > rl.getMinPauseCheckBytes()) {