Reduced logging by BlockManager in default log4j.properties

also fixed a putInto hydra test and update spark module link

cluster/conf/log4j.properties.template

@@ -99,6 +99,7 @@ log4j.logger.org.apache.spark.scheduler.DAGScheduler=WARN
 log4j.logger.org.apache.spark.scheduler.TaskSetManager=WARN
 log4j.logger.org.apache.spark.scheduler.FairSchedulableBuilder=WARN
 log4j.logger.org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend$DriverEndpoint=WARN
+log4j.logger.org.apache.spark.storage.BlockManager=WARN
 log4j.logger.org.apache.spark.storage.BlockManagerInfo=WARN
 log4j.logger.org.apache.hadoop.hive=WARN
 log4j.logger.org.apache.spark.sql.execution.datasources=WARN
@@ -116,13 +117,20 @@ log4j.logger.org.datanucleus=ERROR
 # Task logger created in SparkEnv
 log4j.logger.org.apache.spark.Task=WARN
 log4j.logger.org.apache.spark.sql.catalyst.parser.CatalystSqlParser=WARN
+# HiveExternalCatalog spits out a warning every time a non-hive table is persisted in meta-store
+log4j.logger.org.apache.spark.sql.hive.SnappyHiveExternalCatalog=ERROR
 
 # Keep log-level of some classes as INFO even if root level is higher
 log4j.logger.io.snappydata.impl.LeadImpl=INFO
 log4j.logger.io.snappydata.impl.ServerImpl=INFO
 log4j.logger.io.snappydata.impl.LocatorImpl=INFO
 log4j.logger.spray.can.server.HttpListener=INFO
 
+# Note: all code generation classes that dump using "code" logger should
+# also be listed in ClientSharedUtils.initLog4j for removal in case top-level
+# file has not been set (e.g. common for JDBC clients) else an empty
+# generatedcode.log will be created.
+
 # for generated code of plans
 log4j.logger.org.apache.spark.sql.execution.WholeStageCodegenExec=INFO, code
 log4j.additivity.org.apache.spark.sql.execution.WholeStageCodegenExec=false

cluster/src/main/scala/io/snappydata/impl/LeadImpl.scala

@@ -26,6 +26,7 @@ import scala.collection.JavaConverters._
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.Duration
 import scala.concurrent.{Await, Future}
+
 import akka.actor.ActorSystem
 import com.gemstone.gemfire.CancelException
 import com.gemstone.gemfire.cache.CacheClosedException
@@ -50,15 +51,14 @@ import org.apache.thrift.transport.TTransportException
 import spark.jobserver.JobServer
 import spark.jobserver.auth.{AuthInfo, SnappyAuthenticator, User}
 import spray.routing.authentication.UserPass
+
 import org.apache.spark.sql.collection.{ToolsCallbackInit, Utils}
 import org.apache.spark.sql.execution.SecurityUtils
 import org.apache.spark.sql.hive.thriftserver.SnappyHiveThriftServer2
 import org.apache.spark.sql.{SnappyContext, SnappySession}
 import org.apache.spark.util.LocalDirectoryCleanupUtil
 import org.apache.spark.{Logging, SparkCallbacks, SparkConf, SparkContext, SparkException}
 
-import scala.collection.mutable.ArrayBuffer
-
 class LeadImpl extends ServerImpl with Lead
     with ProtocolOverrides with Logging {
 
@@ -336,10 +336,8 @@ class LeadImpl extends ServerImpl with Lead
 
       // If recovery mode then initialize the recovery service
       if (Misc.getGemFireCache.isSnappyRecoveryMode) {
-        if (enableTableCountInUI.equalsIgnoreCase("true"))
-          RecoveryService.collectViewsAndPrepareCatalog(true)
-        else
-          RecoveryService.collectViewsAndPrepareCatalog(false)
+        RecoveryService.collectViewsAndPrepareCatalog(
+          enableTableCountInUI.equalsIgnoreCase("true"))
       }
 
       if (jobServerWait) {

dtests/src/resources/scripts/putInto/create_and_load_testLTable.sql

@@ -1,4 +1,4 @@
 -- DROP TABLE IF ALREADY EXISTS --
 DROP TABLE IF EXISTS testL;
 
-create table testL (id long, data string, data2 decimal) using column options (partition_by 'id', key_columns 'id');
+create table testL (id long, data string, data2 decimal(38,10)) using column options (partition_by 'id', key_columns 'id');

dtests/src/test/scala/io/snappydata/hydra/putInto/ConcPutIntoTest.scala

@@ -26,10 +26,12 @@ import scala.concurrent.{Await, Future}
 
 object ConcPutIntoTest {
 
-  def concPutInto(primaryLocatorHost: String, primaryLocatorPort: String, numThreads: Integer): Any = {
+  def concPutInto(primaryLocatorHost: String, primaryLocatorPort: String,
+      numThreads: Integer): Any = {
     val globalId = new AtomicInteger()
     val doPut = () => Future {
-      val conn = DriverManager.getConnection("jdbc:snappydata://" + primaryLocatorHost + ":" + primaryLocatorPort)
+      val conn = DriverManager.getConnection(
+        "jdbc:snappydata://" + primaryLocatorHost + ":" + primaryLocatorPort)
       val stmt = conn.createStatement()
       val myId = globalId.getAndIncrement()
       val blockSize = 100000L
@@ -60,7 +62,8 @@ object ConcPutIntoTest {
     queryTasks.foreach(Await.result(_, Duration.Inf))
   }
 
-  def conSelect(primaryLocatorHost: String, primaryLocatorPort: String, numThreads: Integer): Any = {
+  def conSelect(primaryLocatorHost: String, primaryLocatorPort: String,
+      numThreads: Integer): Any = {
     val globalId = new AtomicInteger()
     val doQuery = () => Future {
       val conn = DriverManager.getConnection("jdbc:snappydata://localhost:1527")