package org.apache.zeppelin.interpreter;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URL;

public class SparkDownloadUtils {
  private static Logger LOGGER = LoggerFactory.getLogger(SparkDownloadUtils.class);

  private static String downloadFolder = System.getProperty("user.home") + "/.cache/spark";

  static {
    try {
      FileUtils.forceMkdir(new File(downloadFolder));
    } catch (IOException e) {
      throw new RuntimeException("Fail to create downloadFolder: " + downloadFolder, e);
    }
  }


  public static String downloadSpark(String version) {
    File targetSparkHomeFolder = new File(downloadFolder + "/spark-" + version + "-bin-hadoop2.6");
    if (targetSparkHomeFolder.exists()) {
      LOGGER.info("Skip to download spark as it is already downloaded.");
      return targetSparkHomeFolder.getAbsolutePath();
    }
    // Try mirrors a few times until one succeeds
    for (int i = 0; i < 3; i++) {
      try {
        String preferredMirror = IOUtils.toString(new URL("https://www.apache.org/dyn/closer.lua?preferred=true"));
        File downloadFile = new File(downloadFolder + "/spark-" + version + "-bin-hadoop2.6.tgz");
        String downloadURL = preferredMirror + "/spark/spark-" + version + "/spark-" + version + "-bin-hadoop2.6.tgz";;
        runShellCommand(new String[] {"wget", downloadURL, "-P", downloadFolder});
        runShellCommand(new String[]{"tar", "-xvf", downloadFile.getAbsolutePath(), "-C", downloadFolder});
        break;
      } catch (Exception e) {
        LOGGER.warn("Failed to download Spark", e);
      }
    }
    return targetSparkHomeFolder.getAbsolutePath();
  }

  private static void runShellCommand(String[] commands) throws IOException, InterruptedException {
    LOGGER.info("Starting shell commands: " + StringUtils.join(commands, " "));
    Process process = Runtime.getRuntime().exec(commands);
    StreamGobbler errorGobbler = new StreamGobbler(process.getErrorStream());
    StreamGobbler outputGobbler = new StreamGobbler(process.getInputStream());
    errorGobbler.start();
    outputGobbler.start();
    if (process.waitFor() != 0) {
      throw new IOException("Fail to run shell commands: " + StringUtils.join(commands, " "));
    }
    LOGGER.info("Complete shell commands: " + StringUtils.join(commands, " "));
  }

  private static class StreamGobbler extends Thread {
    InputStream is;

    // reads everything from is until empty.
    StreamGobbler(InputStream is) {
      this.is = is;
    }

    public void run() {
      try {
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        String line = null;
        long startTime = System.currentTimeMillis();
        while ( (line = br.readLine()) != null) {
          // logging per 5 seconds
          if ((System.currentTimeMillis() - startTime) > 5000) {
            LOGGER.info(line);
            startTime = System.currentTimeMillis();
          }
        }
      } catch (IOException ioe) {
        ioe.printStackTrace();
      }
    }
  }
}

@@ -35,16 +35,179 @@
 reter;%0A%0A
+import java.io.BufferedReader;%0Aimport java.io.File;%0Aimport java.io.IOException;%0Aimport java.io.InputStream;%0Aimport java.io.InputStreamReader;%0Aimport java.net.URL;%0A
 import o
@@ -383,172 +383,8 @@
 y;%0A%0A
-import java.io.BufferedReader;%0Aimport java.io.File;%0Aimport java.io.IOException;%0Aimport java.io.InputStream;%0Aimport java.io.InputStreamReader;%0Aimport java.net.URL;%0A%0A
 publ
@@ -1178,16 +1178,48 @@
 ucceeds%0A
+    boolean downloaded = false;%0A
     for 
@@ -1589,17 +1589,16 @@
 .6.tgz%22;
-;
 %0A       
@@ -1773,32 +1773,59 @@
 wnloadFolder%7D);%0A
+        downloaded = true;%0A
         break;%0A 
@@ -1914,24 +1914,803 @@
      %7D%0A    %7D
+%0A%0A    // fallback to use apache archive%0A    // https://archive.apache.org/dist/spark/spark-1.6.3/spark-1.6.3-bin-hadoop2.6.tgz%0A    if (!downloaded) %7B%0A      File downloadFile = new File(downloadFolder + %22/spark-%22 + version + %22-bin-hadoop2.6.tgz%22);%0A      String downloadURL =%0A          %22https://archive.apache.org/dist/spark/spark-%22%0A              + version%0A              + %22/spark-%22%0A              + version%0A              + %22-bin-hadoop2.6.tgz%22;%0A      try %7B%0A        runShellCommand(new String%5B%5D %7B%22wget%22, downloadURL, %22-P%22, downloadFolder%7D);%0A        runShellCommand(%0A            new String%5B%5D %7B%22tar%22, %22-xvf%22, downloadFile.getAbsolutePath(), %22-C%22, downloadFolder%7D);%0A      %7D catch (Exception e) %7B%0A        throw new RuntimeException(%22Fail to download spark %22 + version, e);%0A      %7D%0A    %7D
 %0A    return 

Change Spark download url for spark 1.6.x