Snap 293

cluster/src/dunit/scala/io/snappydata/cluster/ClusterManagerTestBase.scala

@@ -1,3 +1,21 @@
+/*
+ * Copyright (c) 2016 SnappyData, Inc. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License. You
+ * may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License. See accompanying
+ * LICENSE file.
+ */
+
+
 package io.snappydata.cluster
 
 import java.io.File

cluster/src/dunit/scala/io/snappydata/cluster/DynamicJarInstallationDUnitTest.scala

@@ -0,0 +1,176 @@
+/*
+ * Copyright (c) 2016 SnappyData, Inc. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License. You
+ * may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License. See accompanying
+ * LICENSE file.
+ */
+
+
+package io.snappydata.cluster
+
+import java.io.File
+import java.sql.{Connection, DriverManager}
+import io.snappydata.test.dunit.AvailablePortHelper
+import org.apache.spark.sql.SnappyContext
+import org.apache.spark.sql.collection.{Utils => Utility}
+import org.apache.spark.util.SnappyUtils
+
+class DynamicJarInstallationDUnitTest(val s: String)
+    extends ClusterManagerTestBase(s) {
+
+
+  override def tearDown2(): Unit = {
+    Array(vm3, vm2, vm1, vm0).foreach(_.invoke(getClass, "stopNetworkServers"))
+    bootProps.clear()
+  }
+
+  private def getANetConnection(netPort: Int): Connection = {
+    val driver = "com.pivotal.gemfirexd.jdbc.ClientDriver"
+    // scalastyle:off classforname
+    Class.forName(driver).newInstance
+    val url = "jdbc:snappydata://localhost:" + netPort + "/"
+    DriverManager.getConnection(url)
+  }
+
+  def testJarDeployedOnExecutors(): Unit = {
+    val snc = SnappyContext(sc)
+
+    sc.addJar(DynamicJarInstallationDUnitTest.original.getPath)
+
+    // loading Jar class after installing the  jar.
+
+    var countInstances = Utility.mapExecutors(snc,
+      () => {
+        if (DynamicJarInstallationDUnitTest.loadClass("FakeClass1", "1", false)) {
+          Seq(1).iterator
+        } else Iterator.empty
+      }).count
+
+    assert(countInstances == 3)
+
+    // replace the jar file and load again
+
+    val modifiedJarPath = DynamicJarInstallationDUnitTest.modified
+
+    sc.addJar(modifiedJarPath.getPath)
+
+
+    // loading Jar class after installing the  jar
+    countInstances = Utility.mapExecutors(snc,
+      () => {
+        if (DynamicJarInstallationDUnitTest.loadClass("FakeClass1", "1", true)) {
+          Seq(1).iterator
+        } else Iterator.empty
+      }).count
+
+    assert(countInstances == 0)
+
+    // try to verify  another class.
+    countInstances = Utility.mapExecutors(snc,
+      () => {
+        if (DynamicJarInstallationDUnitTest.loadClass("FakeClass2", "2", false)) {
+          Seq(1).iterator
+        } else Iterator.empty
+      }).count
+
+    assert(countInstances == 3)
+  }
+
+  def testJarDeployementwithSnappyshell(): Unit = {
+    val snc = SnappyContext(sc)
+    val sqlJars = SnappyUtils.createJarWithClasses(
+      classNames = Seq("FakeClass4", "FakeClass5", "FakeClass6"),
+      toStringValue = "3" ,
+      Seq.empty, Seq.empty ,
+      "testJar_SNAPPY_JOB_SERVER_JAR_%s.jar".format(System.currentTimeMillis()))
+
+    // test basic operations on snappyContext
+    snc.sql("create table test (x int) using column")
+    snc.dropTable("test")
+
+    val netPort1 = AvailablePortHelper.getRandomAvailableTCPPort
+    vm2.invoke(classOf[ClusterManagerTestBase], "startNetServer", netPort1)
+    val conn = getANetConnection(netPort1)
+
+    val stmt = conn.createStatement()
+
+    stmt.execute("call sqlj.install_jar('" + sqlJars.getPath + "', 'app.sqlJars', 0)")
+
+    // look for jar inside the executors
+    var countInstances = Utility.mapExecutors(snc,
+      () => {
+        if (DynamicJarInstallationDUnitTest.loadClass("FakeClass5", "3")) {
+          Seq(1).iterator
+        } else Iterator.empty
+      }).count
+
+    assert(countInstances == 3)
+
+    // remove the jar and check again
+    stmt.execute("call sqlj.remove_jar('app.sqlJars', 0)")
+
+    // look for jar inside the executors
+    countInstances = Utility.mapExecutors(snc,
+      () => {
+        if (DynamicJarInstallationDUnitTest.loadClass("FakeClass5", "3", true)) {
+          Seq(1).iterator
+        } else Iterator.empty
+      }).count
+
+    assert(countInstances == 0)
+  }
+}
+
+
+object DynamicJarInstallationDUnitTest {
+
+  val original = SnappyUtils.createJarWithClasses(
+    classNames = Seq("FakeClass1", "FakeClass2", "FakeClass3"),
+    toStringValue = "1",
+    Seq.empty, Seq.empty ,
+    "testJar_SNAPPY_JOB_SERVER_JAR_%s.jar".format(System.currentTimeMillis()))
+
+  val modified = SnappyUtils.createJarWithClasses(
+    classNames = Seq("FakeClass2", "FakeClass3", "FakeClass4"),
+    toStringValue = "2",
+    Seq.empty, Seq.empty ,
+    "testJar_SNAPPY_JOB_SERVER_JAR_%s.jar".format(System.currentTimeMillis()))
+
+  @throws[ClassNotFoundException]
+  def loadClass(className: String,
+      version: String = "",
+      catchExpectedException: Boolean = false): Boolean = {
+    val loader = Thread.currentThread().getContextClassLoader
+    assert(loader != null)
+    try {
+      val fakeClass = loader.loadClass(className).newInstance()
+      assert(fakeClass != null)
+      if (!version.isEmpty) assert(fakeClass.toString.equals(version))
+      true
+    } catch {
+      case cnfe: ClassNotFoundException =>
+        if (!catchExpectedException) throw cnfe
+        else false
+    }
+  }
+
+  def getModifiedJar: String = {
+    val oldFile = new File(original.getPath)
+    val newFile = new File(modified.getPath)
+    if (oldFile.exists()) {
+     assert( oldFile.delete() == true)
+    }
+    assert (newFile.renameTo(oldFile) == true)
+    oldFile.getAbsolutePath
+  }
+}

cluster/src/main/java/org/apache/spark/streaming/JavaSnappyStreamingJob.java

@@ -18,28 +18,33 @@
 
 
 import com.typesafe.config.Config;
-import org.apache.spark.sql.JSparkJobValidation;
-import org.apache.spark.sql.JavaJobValidate;
-import org.apache.spark.sql.streaming.SnappyStreamingJob;
+import org.apache.spark.sql.SnappyJobValidate;
+import org.apache.spark.sql.SnappyJobValidation;
 import org.apache.spark.streaming.api.java.JavaSnappyStreamingContext;
+
+import spark.jobserver.SparkJobBase;
 import spark.jobserver.SparkJobValidation;
+import org.apache.spark.util.SnappyUtils;
+import org.apache.spark.util.Utils;
 
-public abstract class JavaSnappyStreamingJob implements SnappyStreamingJob {
+public abstract class JavaSnappyStreamingJob implements SparkJobBase {
 
-  abstract public  Object runJavaJob(JavaSnappyStreamingContext snc, Config jobConfig);
+  abstract public Object runSnappyJob(JavaSnappyStreamingContext snc, Config jobConfig);
 
-  abstract public JSparkJobValidation isValidJob(JavaSnappyStreamingContext snc,
+  abstract public SnappyJobValidation isValidJob(JavaSnappyStreamingContext snc,
       Config jobConfig);
 
   @Override
-  public Object runJob(Object sc, Config jobConfig) {
-    return runJavaJob(new JavaSnappyStreamingContext((SnappyStreamingContext)sc), jobConfig);
+  final public SparkJobValidation validate(Object sc, Config config) {
+    ClassLoader parentLoader = Utils.getContextOrSparkClassLoader();
+    ClassLoader currentLoader = SnappyUtils.getSnappyStoreContextLoader(parentLoader);
+    Thread.currentThread().setContextClassLoader(currentLoader);
+    return SnappyJobValidate.validate(isValidJob(new JavaSnappyStreamingContext((SnappyStreamingContext)sc), config));
   }
 
   @Override
-  public SparkJobValidation validate(Object sc, Config config) {
-    JSparkJobValidation  status =
-        isValidJob(new JavaSnappyStreamingContext((SnappyStreamingContext)sc), config);
-    return JavaJobValidate.validate(status);
+  final public Object runJob(Object sc, Config jobConfig) {
+    return runSnappyJob(new JavaSnappyStreamingContext((SnappyStreamingContext)sc), jobConfig);
   }
-}
+
+}

cluster/src/main/scala/io/snappydata/gemxd/SparkSQLExecuteImpl.scala

@@ -42,8 +42,8 @@ import org.apache.spark.sql.store.CodeGeneration
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{DataFrame, SnappyContext}
 import org.apache.spark.storage.{RDDBlockId, StorageLevel}
+import org.apache.spark.util.{Utils => SparkUtils, SnappyUtils}
 import org.apache.spark.{Logging, SparkContext, SparkEnv}
-
 /**
  * Encapsulates a Spark execution for use in query routing from JDBC.
  */
@@ -58,6 +58,11 @@ class SparkSQLExecuteImpl(val sql: String,
   private[this] val snc = SnappyContextPerConnection
       .getSnappyContextForConnection(ctx.getConnId)
 
+  if (Thread.currentThread().getContextClassLoader != null ) {
+    val loader = SnappyUtils.getSnappyStoreContextLoader(getContextOrCurrentClassLoader)
+    Thread.currentThread().setContextClassLoader(loader)
+  }
+
   snc.setSchema(schema)
 
   private[this] val df: DataFrame = snc.sql(sql)
@@ -237,6 +242,11 @@ class SparkSQLExecuteImpl(val sql: String,
       case _ => (StoredFormatIds.SQL_VARCHAR_ID, -1, -1)
     }
   }
+
+
+  def getContextOrCurrentClassLoader: ClassLoader =
+    Option(Thread.currentThread().getContextClassLoader).getOrElse(getClass.getClassLoader)
+
 }
 
 object SparkSQLExecuteImpl {

cluster/src/main/scala/org/apache/spark/executor/SnappyCoarseGrainedExecutorBackend.scala

@@ -76,4 +76,7 @@ class SnappyCoarseGrainedExecutorBackend(
 
     SparkHadoopUtil.get.stopExecutorDelegationTokenRenewer()
   }
+
+  override def registerExecutor(hostname: String): Executor =
+    new SnappyExecutor(executorId, hostname, env, userClassPath, isLocal = false)
 }

cluster/src/main/scala/org/apache/spark/executor/SnappyExecutor.scala

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2016 SnappyData, Inc. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you
+ * may not use this file except in compliance with the License. You
+ * may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License. See accompanying
+ * LICENSE file.
+ */
+package org.apache.spark.executor
+
+import java.net.URL
+
+import org.apache.spark.SparkEnv
+import org.apache.spark.util.{DynamicURLClassLoader, MutableURLClassLoader}
+
+class SnappyExecutor(
+    executorId: String,
+    executorHostname: String,
+    env: SparkEnv,
+    userClassPath: Seq[URL] = Nil,
+    isLocal: Boolean = false)
+    extends Executor(executorId, executorHostname, env, userClassPath , isLocal) {
+
+  private val overwriteFiles = env.conf.getBoolean("spark.files.overwrite", false)
+
+  override def createClassLoader(urls: Array[URL],
+      parentLoader: ClassLoader, userClassPathFirst: Boolean): MutableURLClassLoader = {
+    new DynamicURLClassLoader(urls, parentLoader, parentFirst = !userClassPathFirst , overwriteFiles)
+  }
+
+}
+
+