This is an automated email from the ASF dual-hosted git repository.

cwylie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git


The following commit(s) were added to refs/heads/master by this push:
     new 9eaf8f5  google-storage: retry GoogleTaskLogs inserts (#6918)
9eaf8f5 is described below

commit 9eaf8f5304e54a65c35395e657cecfafd6f7ecce
Author: David Glasser <glas...@apollographql.com>
AuthorDate: Thu Jan 31 01:21:35 2019 -0800

    google-storage: retry GoogleTaskLogs inserts (#6918)
    
    This is an extension of PR #5750 by @drcrallen which added retry to a 
variety of
    GCS operations, but not to GoogleTaskLogs, which we have found to
    occasionally fail in our cluster.
    
    Also fixes a typo in a variable name and removes an unused private method
    parameter.
    
    Fixes #6912.
---
 .../druid/storage/google/GoogleTaskLogs.java       | 30 +++++++++++++++++-----
 1 file changed, 24 insertions(+), 6 deletions(-)

diff --git 
a/extensions-contrib/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleTaskLogs.java
 
b/extensions-contrib/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleTaskLogs.java
index 0bb80ad..feb373e 100644
--- 
a/extensions-contrib/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleTaskLogs.java
+++ 
b/extensions-contrib/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleTaskLogs.java
@@ -24,6 +24,8 @@ import com.google.common.base.Optional;
 import com.google.common.io.ByteSource;
 import com.google.inject.Inject;
 import org.apache.druid.java.util.common.IOE;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.RetryUtils;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.tasklogs.TaskLogs;
 
@@ -51,7 +53,7 @@ public class GoogleTaskLogs implements TaskLogs
   {
     final String taskKey = getTaskLogKey(taskid);
     LOG.info("Pushing task log %s to: %s", logFile, taskKey);
-    pushTaskFile(taskid, logFile, taskKey);
+    pushTaskFile(logFile, taskKey);
   }
 
   @Override
@@ -59,17 +61,33 @@ public class GoogleTaskLogs implements TaskLogs
   {
     final String taskKey = getTaskReportKey(taskid);
     LOG.info("Pushing task reports %s to: %s", reportFile, taskKey);
-    pushTaskFile(taskid, reportFile, taskKey);
+    pushTaskFile(reportFile, taskKey);
   }
 
-  private void pushTaskFile(final String taskid, final File logFile, final 
String taskKey) throws IOException
+  private void pushTaskFile(final File logFile, final String taskKey) throws 
IOException
   {
-    FileInputStream fileSteam = new FileInputStream(logFile);
+    FileInputStream fileStream = new FileInputStream(logFile);
 
-    InputStreamContent mediaContent = new InputStreamContent("text/plain", 
fileSteam);
+    InputStreamContent mediaContent = new InputStreamContent("text/plain", 
fileStream);
     mediaContent.setLength(logFile.length());
 
-    storage.insert(config.getBucket(), taskKey, mediaContent);
+    try {
+      RetryUtils.retry(
+          (RetryUtils.Task<Void>) () -> {
+            storage.insert(config.getBucket(), taskKey, mediaContent);
+            return null;
+          },
+          GoogleUtils::isRetryable,
+          1,
+          5
+      );
+    }
+    catch (IOException e) {
+      throw e;
+    }
+    catch (Exception e) {
+      throw new RE(e, "Failed to upload [%s] to [%s]", logFile, taskKey);
+    }
   }
 
   @Override


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org

Reply via email to