feat: add configuration for meta query timeout (#103)

XiaoMi · May 12, 2020 · 76620ab · 76620ab
1 parent 6a76f58
commit 76620ab
Show file tree

Hide file tree

Showing 5 changed files with 56 additions and 7 deletions.
diff --git a/configuration/pegasus.properties b/configuration/pegasus.properties
@@ -4,3 +4,4 @@ async_workers = 4
 enable_perf_counter = false
 perf_counter_tags = cluster=onebox,app=unit_test
 push_counter_interval_secs = 10
+meta_query_timeout = 5000
diff --git a/src/main/java/com/xiaomi/infra/pegasus/client/ClientOptions.java b/src/main/java/com/xiaomi/infra/pegasus/client/ClientOptions.java
@@ -25,6 +25,7 @@
  *          .enablePerfCounter(false)
  *          .falconPerfCounterTags("")
  *          .falconPushInterval(Duration.ofSeconds(10))
+ *          .metaQueryTimeout(Duration.ofMillis(5000))
  *          .build();
  * }</pre>
  */
@@ -38,6 +39,7 @@ public class ClientOptions {
   public static final String DEFAULT_FALCON_PERF_COUNTER_TAGS = "";
   public static final Duration DEFAULT_FALCON_PUSH_INTERVAL = Duration.ofSeconds(10);
   public static final boolean DEFAULT_ENABLE_WRITE_LIMIT = true;
+  public static final Duration DEFAULT_META_QUERY_TIMEOUT = Duration.ofMillis(5000);
 
   private final String metaServers;
   private final Duration operationTimeout;
@@ -46,6 +48,7 @@ public class ClientOptions {
   private final String falconPerfCounterTags;
   private final Duration falconPushInterval;
   private final boolean enableWriteLimit;
+  private final Duration metaQueryTimeout;
 
   protected ClientOptions(Builder builder) {
     this.metaServers = builder.metaServers;
@@ -55,6 +58,7 @@ protected ClientOptions(Builder builder) {
     this.falconPerfCounterTags = builder.falconPerfCounterTags;
     this.falconPushInterval = builder.falconPushInterval;
     this.enableWriteLimit = builder.enableWriteLimit;
+    this.metaQueryTimeout = builder.metaQueryTimeout;
   }
 
   protected ClientOptions(ClientOptions original) {
@@ -65,6 +69,7 @@ protected ClientOptions(ClientOptions original) {
     this.falconPerfCounterTags = original.getFalconPerfCounterTags();
     this.falconPushInterval = original.getFalconPushInterval();
     this.enableWriteLimit = original.isWriteLimitEnabled();
+    this.metaQueryTimeout = original.getMetaQueryTimeout();
   }
 
   /**
@@ -108,7 +113,8 @@ public boolean equals(Object options) {
           && this.enablePerfCounter == clientOptions.enablePerfCounter
           && this.falconPerfCounterTags.equals(clientOptions.falconPerfCounterTags)
           && this.falconPushInterval.toMillis() == clientOptions.falconPushInterval.toMillis()
-          && this.enableWriteLimit == clientOptions.enableWriteLimit;
+          && this.enableWriteLimit == clientOptions.enableWriteLimit
+          && this.metaQueryTimeout.toMillis() == clientOptions.metaQueryTimeout.toMillis();
     }
     return false;
   }
@@ -132,6 +138,8 @@ public String toString() {
         + falconPushInterval.getSeconds()
         + ",enableWriteLimit="
         + enableWriteLimit
+        + ", metaQueryTimeout(ms)="
+        + metaQueryTimeout.toMillis()
         + '}';
   }
 
@@ -144,6 +152,7 @@ public static class Builder {
     private String falconPerfCounterTags = DEFAULT_FALCON_PERF_COUNTER_TAGS;
     private Duration falconPushInterval = DEFAULT_FALCON_PUSH_INTERVAL;
     private boolean enableWriteLimit = DEFAULT_ENABLE_WRITE_LIMIT;
+    private Duration metaQueryTimeout = DEFAULT_META_QUERY_TIMEOUT;
 
     protected Builder() {}
 
@@ -234,6 +243,18 @@ public Builder enableWriteLimit(boolean enableWriteLimit) {
       return this;
     }
 
+    /**
+     * The timeout for query meta server. Defaults to {@literal 5000ms}, see {@link
+     * #DEFAULT_META_QUERY_TIMEOUT}.
+     *
+     * @param metaQueryTimeout metaQueryTimeout
+     * @return {@code this}
+     */
+    public Builder metaQueryTimeout(Duration metaQueryTimeout) {
+      this.metaQueryTimeout = metaQueryTimeout;
+      return this;
+    }
+
     /**
      * Create a new instance of {@link ClientOptions}.
      *
@@ -260,7 +281,8 @@ public ClientOptions.Builder mutate() {
         .enablePerfCounter(isEnablePerfCounter())
         .falconPerfCounterTags(getFalconPerfCounterTags())
         .falconPushInterval(getFalconPushInterval())
-        .enableWriteLimit(isWriteLimitEnabled());
+        .enableWriteLimit(isWriteLimitEnabled())
+        .metaQueryTimeout(getMetaQueryTimeout());
     return builder;
   }
 
@@ -331,4 +353,13 @@ public Duration getFalconPushInterval() {
   public boolean isWriteLimitEnabled() {
     return enableWriteLimit;
   }
+
+  /**
+   * The timeout for query meta server. Defaults to {@literal 5000ms}.
+   *
+   * @return the timeout for query meta server.
+   */
+  public Duration getMetaQueryTimeout() {
+    return metaQueryTimeout;
+  }
 }
diff --git a/src/main/java/com/xiaomi/infra/pegasus/client/PegasusClientFactory.java b/src/main/java/com/xiaomi/infra/pegasus/client/PegasusClientFactory.java
@@ -55,6 +55,8 @@ public static PegasusClientInterface createClient(ClientOptions options) throws
     pegasusConfig.setProperty(
         "push_counter_interval_secs", String.valueOf(options.getFalconPushInterval().getSeconds()));
     pegasusConfig.setProperty("enable_write_limit", String.valueOf(options.isWriteLimitEnabled()));
+    pegasusConfig.setProperty(
+        "meta_query_timeout", String.valueOf(options.getMetaQueryTimeout().toMillis()));
     return new PegasusClient(pegasusConfig);
   }
 

diff --git a/src/main/java/com/xiaomi/infra/pegasus/rpc/ClusterOptions.java b/src/main/java/com/xiaomi/infra/pegasus/rpc/ClusterOptions.java
@@ -27,14 +27,18 @@ public class ClusterOptions {
   public static final String PEGASUS_PUSH_COUNTER_INTERVAL_SECS_KEY = "push_counter_interval_secs";
   public static final String PEGASUS_PUSH_COUNTER_INTERVAL_SECS_DEF = "60";
 
+  public static final String PEGASUS_META_QUERY_TIMEOUT_KEY = "meta_query_timeout";
+  public static final String PEGASUS_META_QUERY_TIMEOUT_DEF = "5000";
+
   public static String[] allKeys() {
     return new String[] {
       PEGASUS_META_SERVERS_KEY,
       PEGASUS_OPERATION_TIMEOUT_KEY,
       PEGASUS_ASYNC_WORKERS_KEY,
       PEGASUS_ENABLE_PERF_COUNTER_KEY,
       PEGASUS_PERF_COUNTER_TAGS_KEY,
-      PEGASUS_PUSH_COUNTER_INTERVAL_SECS_KEY
+      PEGASUS_PUSH_COUNTER_INTERVAL_SECS_KEY,
+      PEGASUS_META_QUERY_TIMEOUT_KEY
     };
   }
 
@@ -44,6 +48,7 @@ public static String[] allKeys() {
   private final boolean enablePerfCounter;
   private final String perfCounterTags;
   private final int pushCounterIntervalSecs;
+  private final int metaQueryTimeout;
 
   public int operationTimeout() {
     return this.operationTimeout;
@@ -69,6 +74,10 @@ public int pushCounterIntervalSecs() {
     return this.pushCounterIntervalSecs;
   }
 
+  public int metaQueryTimeout() {
+    return this.metaQueryTimeout;
+  }
+
   public static ClusterOptions create(Properties config) {
     int operationTimeout =
         Integer.parseInt(
@@ -96,18 +105,22 @@ public static ClusterOptions create(Properties config) {
         Integer.parseInt(
             config.getProperty(
                 PEGASUS_PUSH_COUNTER_INTERVAL_SECS_KEY, PEGASUS_PUSH_COUNTER_INTERVAL_SECS_DEF));
+    int metaQueryTimeout =
+        Integer.parseInt(
+            config.getProperty(PEGASUS_META_QUERY_TIMEOUT_KEY, PEGASUS_META_QUERY_TIMEOUT_DEF));
 
     return new ClusterOptions(
         operationTimeout,
         address,
         asyncWorkers,
         enablePerfCounter,
         perfCounterTags,
-        pushIntervalSecs);
+        pushIntervalSecs,
+        metaQueryTimeout);
   }
 
   public static ClusterOptions forTest(String[] metaList) {
-    return new ClusterOptions(1000, metaList, 1, false, null, 60);
+    return new ClusterOptions(1000, metaList, 1, false, null, 60, 1000);
   }
 
   private ClusterOptions(
@@ -116,12 +129,14 @@ private ClusterOptions(
       int asyncWorkers,
       boolean enablePerfCounter,
       String perfCounterTags,
-      int pushCounterIntervalSecs) {
+      int pushCounterIntervalSecs,
+      int metaQueryTimeout) {
     this.operationTimeout = operationTimeout;
     this.metaList = metaList;
     this.asyncWorkers = asyncWorkers;
     this.enablePerfCounter = enablePerfCounter;
     this.perfCounterTags = perfCounterTags;
     this.pushCounterIntervalSecs = pushCounterIntervalSecs;
+    this.metaQueryTimeout = metaQueryTimeout;
   }
 }
diff --git a/src/main/java/com/xiaomi/infra/pegasus/rpc/async/ClusterManager.java b/src/main/java/com/xiaomi/infra/pegasus/rpc/async/ClusterManager.java
@@ -58,7 +58,7 @@ public ClusterManager(ClusterOptions opts) throws IllegalArgumentException {
     metaList = opts.metaList();
     // the constructor of meta session is depend on the replicaSessions,
     // so the replicaSessions should be initialized earlier
-    metaSession = new MetaSession(this, opts.metaList(), opts.operationTimeout(), 10, metaGroup);
+    metaSession = new MetaSession(this, opts.metaList(), opts.metaQueryTimeout(), 10, metaGroup);
   }
 
   public EventExecutor getExecutor() {