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