11617

Build # 11617

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/3f465074fbb0d9a4d1153345d37aece093ce2357">3f465074f">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/3f465074fbb0d9a4d1153345d37aece093ce2357">3f465074f</a><a href="https://github.com/apache/carbondata/commit/3f465074fbb0d9a4d1153345d37aece093ce2357">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/3f465074fbb0d9a4d1153345d37aece093ce2357&quot;&gt;3f465074f&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/3f465074fbb0d9a4d1153345d37aece093ce2357&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

211 of 211 new or added lines in 10 files covered. (100.0%)

64264 of 82243 relevant lines covered (78.14%)

1.03 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.89

/integration/spark-common/src/main/scala/org/apache/spark/sql/hive/DistributionUtil.scala

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.
 */

package org.apache.spark.sql.hive

import java.net.{InetAddress, InterfaceAddress, NetworkInterface}

import scala.collection.JavaConverters._
import scala.util.control.Breaks._

import org.apache.spark.SparkContext
import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.block.Distributable
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.processing.util.CarbonLoaderUtil

object DistributionUtil {
  @transient
  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
  /*
   *  minimum required registered resource for starting block distribution
   */
  lazy val minRegisteredResourceRatio: Double = {
    val value: String = CarbonProperties.getInstance()
      .getProperty(CarbonCommonConstants.CARBON_SCHEDULER_MIN_REGISTERED_RESOURCES_RATIO,
        CarbonCommonConstants.CARBON_SCHEDULER_MIN_REGISTERED_RESOURCES_RATIO_DEFAULT)
    java.lang.Double.parseDouble(value)
  }

  /*
   * node registration wait time
   */
  lazy val dynamicAllocationSchTimeOut: Integer = {
    val value: String = CarbonProperties.getInstance()
      .getProperty(CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_TIMEOUT,
        CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_TIMEOUT_DEFAULT)
    // milli second
    java.lang.Integer.parseInt(value) * 1000
  }

  /*
   * This method will return the list of executers in the cluster.
   * For this we take the  memory status of all node with getExecutorMemoryStatus
   * and extract the keys. getExecutorMemoryStatus also returns the driver memory also
   * In client mode driver will run in the localhost
   * There can be executor spawn in same drive node. So we can remove first occurance of
   * localhost for retriving executor list
   */
  def getNodeList(sparkContext: SparkContext): Array[String] = {
    val arr = sparkContext.getExecutorMemoryStatus.map { kv =>
      kv._1.split(":")(0)
    }.toSeq
    val localhostIPs = getLocalhostIPs
    val selectedLocalIPList = localhostIPs.filter(arr.contains(_))

    val nodelist: List[String] = withoutDriverIP(arr.toList)(selectedLocalIPList.contains(_))
    val masterMode = sparkContext.getConf.get("spark.master")
    if (nodelist.nonEmpty) {
      // Specific for Yarn Mode
      if ("yarn-cluster".equals(masterMode) || "yarn-client".equals(masterMode)) {
        val nodeNames = nodelist.map { x =>
          val addr = InetAddress.getByName(x)
          addr.getHostName
        }
        nodeNames.toArray
      } else {
        // For Standalone cluster, node IPs will be returned.
        nodelist.toArray
      }
    } else {
      Seq(InetAddress.getLocalHost.getHostName).toArray
    }
  }

  private def getLocalhostIPs = {
    val iface = NetworkInterface.getNetworkInterfaces
    var addresses: List[InterfaceAddress] = List.empty
    while (iface.hasMoreElements) {
      addresses = iface.nextElement().getInterfaceAddresses.asScala.toList ++ addresses
    }
    val inets = addresses.map(_.getAddress.getHostName)
    inets
  }

  /*
   * This method will remove the first occurance of any of the ips  mentioned in the predicate.
   * Eg: l = List(Master,slave1,Master,slave2,slave3) is the list of nodes where first Master is
   * the Driver  node.
   * this method withoutFirst (l)(x=> x == 'Master') will remove the first occurance of Master.
   * The resulting List containt List(slave1,Master,slave2,slave3)
   */
  def withoutDriverIP[A](xs: List[A])(p: A => Boolean): List[A] = {
    xs match {
      case x :: rest => if (p(x)) {
        rest
      } else {
        x :: withoutDriverIP(rest)(p)
      }
      case _ => Nil
    }
  }

  /**
   *
   * Checking if the existing executors is greater than configured executors, if yes
   * returning configured executors.
   *
   * @param blockList total number of blocks in the identified segments
   * @param sparkContext
   * @return
   */
  def ensureExecutorsAndGetNodeList(blockList: Seq[Distributable],
      sparkContext: SparkContext): Seq[String] = {
    val nodeMapping = CarbonLoaderUtil.nodeBlockMapping(blockList.asJava)
    ensureExecutorsByNumberAndGetNodeList(nodeMapping, blockList, sparkContext)
  }

  def ensureExecutorsByNumberAndGetNodeList(nodesOfData: Int,
      sparkContext: SparkContext): Seq[String] = {
    val confExecutors: Int = getConfiguredExecutors(sparkContext)
    LOGGER.info(s"Executors configured : $confExecutors")
    val requiredExecutors = if (nodesOfData < 1 || nodesOfData > confExecutors) {
      confExecutors
    } else {
      nodesOfData
    }
    // request for starting the number of required executors
    ensureExecutors(sparkContext, requiredExecutors)
    getDistinctNodesList(sparkContext, requiredExecutors)
  }

  /**
   * This method will ensure that the required/configured number of executors are requested
   * for processing the identified blocks
   *
   * @param nodeMapping
   * @param blockList
   * @param sparkContext
   * @return
   */
  private def ensureExecutorsByNumberAndGetNodeList(
      nodeMapping: java.util.Map[String, java.util.List[Distributable]],
      blockList: Seq[Distributable],
      sparkContext: SparkContext): Seq[String] = {
    val nodesOfData = nodeMapping.size()
    val confExecutors: Int = getConfiguredExecutors(sparkContext)
    LOGGER.info(s"Executors configured : $confExecutors")
    val requiredExecutors = if (nodesOfData < 1) {
      1
    } else if (nodesOfData > confExecutors) {
      confExecutors
    } else if (confExecutors > nodesOfData) {
      var totalExecutorsToBeRequested = nodesOfData
      // If total number of blocks are greater than the nodes identified then ensure
      // that the configured number of max executors can be opened based on the difference of
      // block list size and nodes identified
      if (blockList.size > nodesOfData) {
        // e.g 1. blockList size = 40, confExecutors = 6, then all 6 executors
        // need to be opened
        // 2. blockList size = 4, confExecutors = 6, then
        // total 4 executors need to be opened
        if (blockList.size > confExecutors) {
          totalExecutorsToBeRequested = confExecutors
        } else {
          totalExecutorsToBeRequested = blockList.size
        }
      }
      LOGGER.info(s"Total executors requested: $totalExecutorsToBeRequested")
      totalExecutorsToBeRequested
    } else {
      nodesOfData
    }
    // request for starting the number of required executors
    ensureExecutors(sparkContext, requiredExecutors, blockList.size)
    getDistinctNodesList(sparkContext, requiredExecutors)
  }

  /**
   * This method will return the configured executors
   *
   * @param sparkContext
   * @return
   */
  private def getConfiguredExecutors(sparkContext: SparkContext): Int = {
    var confExecutors: Int = 0
    if (sparkContext.getConf.getBoolean("spark.dynamicAllocation.enabled", false)) {
      // default value for spark.dynamicAllocation.maxExecutors is infinity
      confExecutors = sparkContext.getConf.getInt("spark.dynamicAllocation.maxExecutors", 1)
      LOGGER.info(s"spark.dynamicAllocation.maxExecutors property is set to = $confExecutors")
    } else {
      // default value for spark.executor.instances is 2
      confExecutors = sparkContext.getConf.getInt("spark.executor.instances", 1)
      LOGGER.info(s"spark.executor.instances property is set to = $confExecutors")
    }
    confExecutors
  }

  /**
   * This method will return the distinct nodes list
   *
   * @param sparkContext
   * @param requiredExecutors
   * @return
   */
  private def getDistinctNodesList(sparkContext: SparkContext,
      requiredExecutors: Int): Seq[String] = {
    val startTime = System.currentTimeMillis()
    var nodes = DistributionUtil.getNodeList(sparkContext)
    // calculate the number of times loop has to run to check for starting
    // the requested number of executors
    val threadSleepTime =
    CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME
    val maxRetryCount = calculateMaxRetry
    var maxTimes = maxRetryCount
    breakable {
      while (nodes.length < requiredExecutors && maxTimes > 0) {
        Thread.sleep(threadSleepTime);
        nodes = DistributionUtil.getNodeList(sparkContext)
        maxTimes = maxTimes - 1;
        val resourceRatio = (nodes.length.toDouble / requiredExecutors)
        if (resourceRatio.compareTo(minRegisteredResourceRatio) >= 0) {
          break
        }
      }
    }
    val timDiff = System.currentTimeMillis() - startTime
    LOGGER.info(s"Total Time taken to ensure the required executors : $timDiff")
    LOGGER.info(s"Time elapsed to allocate the required executors: " +
      s"${(maxRetryCount - maxTimes) * threadSleepTime}")
    nodes.distinct.toSeq
  }

  /**
   * Requesting the extra executors other than the existing ones.
   *
   * @param sc                   sparkContext
   * @param requiredExecutors         required number of executors to be requested
   * @param localityAwareTasks   The number of pending tasks which is locality required
   * @param hostToLocalTaskCount A map to store hostname with its possible task number running on it
   * @return
   */
  def ensureExecutors(sc: SparkContext,
      requiredExecutors: Int,
      localityAwareTasks: Int = 0,
      hostToLocalTaskCount: Map[String, Int] = Map.empty): Boolean = {
    sc.schedulerBackend match {
      case b: CoarseGrainedSchedulerBackend =>
        if (requiredExecutors > 0) {
          LOGGER.info(s"Requesting total executors: $requiredExecutors")
          b.requestTotalExecutors(requiredExecutors, localityAwareTasks, hostToLocalTaskCount)
        }
        true
      case _ =>
        false
    }
  }

  /**
   * This method will calculate how many times a loop will run with an interval of given sleep
   * time to wait for requested executors to come up
    *
    * @return The max retry count
   */
  def calculateMaxRetry(): Int = {
    val remainder = dynamicAllocationSchTimeOut % CarbonCommonConstants
      .CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME
    val retryCount: Int = dynamicAllocationSchTimeOut / CarbonCommonConstants
      .CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME
    if (remainder > 0) {
      retryCount + 1
    } else {
      retryCount
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.sql.hive
19
20	import java.net.{InetAddress, InterfaceAddress, NetworkInterface}
21
22	import scala.collection.JavaConverters._
23	import scala.util.control.Breaks._
24
25	import org.apache.spark.SparkContext
26	import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
27
28	import org.apache.carbondata.common.logging.LogServiceFactory
29	import org.apache.carbondata.core.constants.CarbonCommonConstants
30	import org.apache.carbondata.core.datastore.block.Distributable
31	import org.apache.carbondata.core.util.CarbonProperties
32	import org.apache.carbondata.processing.util.CarbonLoaderUtil
33
34	object DistributionUtil {
35	@transient
36	val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	1✔
37	/*
38	* minimum required registered resource for starting block distribution
39	*/
40	lazy val minRegisteredResourceRatio: Double = {	×
41	val value: String = CarbonProperties.getInstance()	×
42	.getProperty(CarbonCommonConstants.CARBON_SCHEDULER_MIN_REGISTERED_RESOURCES_RATIO,	×
43	CarbonCommonConstants.CARBON_SCHEDULER_MIN_REGISTERED_RESOURCES_RATIO_DEFAULT)	×
44	java.lang.Double.parseDouble(value)	×
45	}
46
47	/*
48	* node registration wait time
49	*/
50	lazy val dynamicAllocationSchTimeOut: Integer = {	1✔
51	val value: String = CarbonProperties.getInstance()	1✔
52	.getProperty(CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_TIMEOUT,	1✔
53	CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_TIMEOUT_DEFAULT)	1✔
54	// milli second
55	java.lang.Integer.parseInt(value) * 1000	1✔
56	}
57
58	/*
59	* This method will return the list of executers in the cluster.
60	* For this we take the memory status of all node with getExecutorMemoryStatus
61	* and extract the keys. getExecutorMemoryStatus also returns the driver memory also
62	* In client mode driver will run in the localhost
63	* There can be executor spawn in same drive node. So we can remove first occurance of
64	* localhost for retriving executor list
65	*/
66	def getNodeList(sparkContext: SparkContext): Array[String] = {	1✔
67	val arr = sparkContext.getExecutorMemoryStatus.map { kv =>	1✔
68	kv._1.split(":")(0)	1✔
69	}.toSeq	1✔
70	val localhostIPs = getLocalhostIPs	1✔
71	val selectedLocalIPList = localhostIPs.filter(arr.contains(_))	1✔
72
73	val nodelist: List[String] = withoutDriverIP(arr.toList)(selectedLocalIPList.contains(_))	1✔
74	val masterMode = sparkContext.getConf.get("spark.master")	1✔
75	if (nodelist.nonEmpty) {	1✔
76	// Specific for Yarn Mode
77	if ("yarn-cluster".equals(masterMode) \|\| "yarn-client".equals(masterMode)) {	×
78	val nodeNames = nodelist.map { x =>	×
79	val addr = InetAddress.getByName(x)	×
80	addr.getHostName	×
81	}
82	nodeNames.toArray	×
83	} else {
84	// For Standalone cluster, node IPs will be returned.
85	nodelist.toArray	×
86	}
87	} else {
88	Seq(InetAddress.getLocalHost.getHostName).toArray	1✔
89	}
90	}
91
92	private def getLocalhostIPs = {
93	val iface = NetworkInterface.getNetworkInterfaces	1✔
94	var addresses: List[InterfaceAddress] = List.empty	1✔
95	while (iface.hasMoreElements) {	1✔
96	addresses = iface.nextElement().getInterfaceAddresses.asScala.toList ++ addresses	1✔
97	}
98	val inets = addresses.map(_.getAddress.getHostName)	1✔
99	inets	1✔
100	}
101
102	/*
103	* This method will remove the first occurance of any of the ips mentioned in the predicate.
104	* Eg: l = List(Master,slave1,Master,slave2,slave3) is the list of nodes where first Master is
105	* the Driver node.
106	* this method withoutFirst (l)(x=> x == 'Master') will remove the first occurance of Master.
107	* The resulting List containt List(slave1,Master,slave2,slave3)
108	*/
109	def withoutDriverIP[A](xs: List[A])(p: A => Boolean): List[A] = {
110	xs match {	1✔
111	case x :: rest => if (p(x)) {	1✔
112	rest	1✔
113	} else {
114	x :: withoutDriverIP(rest)(p)	×
115	}
116	case _ => Nil	×
117	}
118	}
119
120	/**
121	*
122	* Checking if the existing executors is greater than configured executors, if yes
123	* returning configured executors.
124	*
125	* @param blockList total number of blocks in the identified segments
126	* @param sparkContext
127	* @return
128	*/
129	def ensureExecutorsAndGetNodeList(blockList: Seq[Distributable],
130	sparkContext: SparkContext): Seq[String] = {
131	val nodeMapping = CarbonLoaderUtil.nodeBlockMapping(blockList.asJava)	1✔
132	ensureExecutorsByNumberAndGetNodeList(nodeMapping, blockList, sparkContext)	1✔
133	}
134
135	def ensureExecutorsByNumberAndGetNodeList(nodesOfData: Int,
136	sparkContext: SparkContext): Seq[String] = {
137	val confExecutors: Int = getConfiguredExecutors(sparkContext)	1✔
138	LOGGER.info(s"Executors configured : $confExecutors")	1✔
139	val requiredExecutors = if (nodesOfData < 1 \|\| nodesOfData > confExecutors) {	1✔
140	confExecutors	1✔
141	} else {
142	nodesOfData	1✔
143	}
144	// request for starting the number of required executors
145	ensureExecutors(sparkContext, requiredExecutors)	1✔
146	getDistinctNodesList(sparkContext, requiredExecutors)	1✔
147	}
148
149	/**
150	* This method will ensure that the required/configured number of executors are requested
151	* for processing the identified blocks
152	*
153	* @param nodeMapping
154	* @param blockList
155	* @param sparkContext
156	* @return
157	*/
158	private def ensureExecutorsByNumberAndGetNodeList(
159	nodeMapping: java.util.Map[String, java.util.List[Distributable]],
160	blockList: Seq[Distributable],
161	sparkContext: SparkContext): Seq[String] = {
162	val nodesOfData = nodeMapping.size()	1✔
163	val confExecutors: Int = getConfiguredExecutors(sparkContext)	1✔
164	LOGGER.info(s"Executors configured : $confExecutors")	1✔
165	val requiredExecutors = if (nodesOfData < 1) {	1✔
166	1	×
167	} else if (nodesOfData > confExecutors) {	1✔
168	confExecutors	×
169	} else if (confExecutors > nodesOfData) {	1✔
170	var totalExecutorsToBeRequested = nodesOfData	×
171	// If total number of blocks are greater than the nodes identified then ensure
172	// that the configured number of max executors can be opened based on the difference of
173	// block list size and nodes identified
174	if (blockList.size > nodesOfData) {	×
175	// e.g 1. blockList size = 40, confExecutors = 6, then all 6 executors
176	// need to be opened
177	// 2. blockList size = 4, confExecutors = 6, then
178	// total 4 executors need to be opened
179	if (blockList.size > confExecutors) {	×
180	totalExecutorsToBeRequested = confExecutors	×
181	} else {
182	totalExecutorsToBeRequested = blockList.size	×
183	}
184	}
185	LOGGER.info(s"Total executors requested: $totalExecutorsToBeRequested")	×
186	totalExecutorsToBeRequested	×
187	} else {
188	nodesOfData	1✔
189	}
190	// request for starting the number of required executors
191	ensureExecutors(sparkContext, requiredExecutors, blockList.size)	1✔
192	getDistinctNodesList(sparkContext, requiredExecutors)	1✔
193	}
194
195	/**
196	* This method will return the configured executors
197	*
198	* @param sparkContext
199	* @return
200	*/
201	private def getConfiguredExecutors(sparkContext: SparkContext): Int = {
202	var confExecutors: Int = 0	1✔
203	if (sparkContext.getConf.getBoolean("spark.dynamicAllocation.enabled", false)) {	1✔
204	// default value for spark.dynamicAllocation.maxExecutors is infinity
205	confExecutors = sparkContext.getConf.getInt("spark.dynamicAllocation.maxExecutors", 1)	×
206	LOGGER.info(s"spark.dynamicAllocation.maxExecutors property is set to = $confExecutors")	×
207	} else {
208	// default value for spark.executor.instances is 2
209	confExecutors = sparkContext.getConf.getInt("spark.executor.instances", 1)	1✔
210	LOGGER.info(s"spark.executor.instances property is set to = $confExecutors")	1✔
211	}
212	confExecutors	1✔
213	}
214
215	/**
216	* This method will return the distinct nodes list
217	*
218	* @param sparkContext
219	* @param requiredExecutors
220	* @return
221	*/
222	private def getDistinctNodesList(sparkContext: SparkContext,
223	requiredExecutors: Int): Seq[String] = {
224	val startTime = System.currentTimeMillis()	1✔
225	var nodes = DistributionUtil.getNodeList(sparkContext)	1✔
226	// calculate the number of times loop has to run to check for starting
227	// the requested number of executors
228	val threadSleepTime =	1✔
229	CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME	1✔
230	val maxRetryCount = calculateMaxRetry	1✔
231	var maxTimes = maxRetryCount	1✔
232	breakable {	1✔
233	while (nodes.length < requiredExecutors && maxTimes > 0) {	1✔
234	Thread.sleep(threadSleepTime);	×
235	nodes = DistributionUtil.getNodeList(sparkContext)	×
236	maxTimes = maxTimes - 1;	×
237	val resourceRatio = (nodes.length.toDouble / requiredExecutors)	×
238	if (resourceRatio.compareTo(minRegisteredResourceRatio) >= 0) {	×
239	break	×
240	}
241	}
242	}
243	val timDiff = System.currentTimeMillis() - startTime	1✔
244	LOGGER.info(s"Total Time taken to ensure the required executors : $timDiff")	1✔
245	LOGGER.info(s"Time elapsed to allocate the required executors: " +	1✔
246	s"${(maxRetryCount - maxTimes) * threadSleepTime}")	1✔
247	nodes.distinct.toSeq	1✔
248	}
249
250	/**
251	* Requesting the extra executors other than the existing ones.
252	*
253	* @param sc sparkContext
254	* @param requiredExecutors required number of executors to be requested
255	* @param localityAwareTasks The number of pending tasks which is locality required
256	* @param hostToLocalTaskCount A map to store hostname with its possible task number running on it
257	* @return
258	*/
259	def ensureExecutors(sc: SparkContext,
260	requiredExecutors: Int,
261	localityAwareTasks: Int = 0,	1✔
262	hostToLocalTaskCount: Map[String, Int] = Map.empty): Boolean = {	1✔
263	sc.schedulerBackend match {	1✔
264	case b: CoarseGrainedSchedulerBackend =>	1✔
265	if (requiredExecutors > 0) {	×
266	LOGGER.info(s"Requesting total executors: $requiredExecutors")	×
267	b.requestTotalExecutors(requiredExecutors, localityAwareTasks, hostToLocalTaskCount)	×
268	}
269	true	×
270	case _ =>
271	false	1✔
272	}
273	}
274
275	/**
276	* This method will calculate how many times a loop will run with an interval of given sleep
277	* time to wait for requested executors to come up
278	*
279	* @return The max retry count
280	*/
281	def calculateMaxRetry(): Int = {	1✔
282	val remainder = dynamicAllocationSchTimeOut % CarbonCommonConstants	1✔
283	.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME	1✔
284	val retryCount: Int = dynamicAllocationSchTimeOut / CarbonCommonConstants	1✔
285	.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME	1✔
286	if (remainder > 0) {	1✔
287	retryCount + 1	×
288	} else {
289	retryCount	1✔
290	}
291	}
292	}	1✔

apache / carbondata / 11617

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous