You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/02/03 12:07:03 UTC
[incubator-celeborn] branch main updated: [CELEBORN-269][BUG] Disable replication throw NPE when removeBatch in pushDataHandler (#1203)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 2e689128 [CELEBORN-269][BUG] Disable replication throw NPE when removeBatch in pushDataHandler (#1203)
2e689128 is described below
commit 2e689128123632ebd067cecca920ece1b6804736
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Fri Feb 3 20:06:59 2023 +0800
[CELEBORN-269][BUG] Disable replication throw NPE when removeBatch in pushDataHandler (#1203)
---
.../celeborn/tests/spark/PushDataTimeoutTest.scala | 46 ++++++++++++----------
.../service/deploy/worker/PushDataHandler.scala | 31 +++++++++------
2 files changed, 44 insertions(+), 33 deletions(-)
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
index 87a30c5e..d7cab66f 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
@@ -46,30 +46,34 @@ class PushDataTimeoutTest extends AnyFunSuite
}
test("celeborn spark integration test - pushdata timeout") {
- val sparkConf = new SparkConf().setAppName("rss-demo").setMaster("local[4]")
- .set("spark.celeborn.push.data.timeout", "5s")
- val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
- val combineResult = combine(sparkSession)
- val groupbyResult = groupBy(sparkSession)
- val repartitionResult = repartition(sparkSession)
- val sqlResult = runsql(sparkSession)
+ Seq("false", "true").foreach { enabled =>
+ val sparkConf = new SparkConf().setAppName("rss-demo").setMaster("local[4]")
+ .set("spark.celeborn.push.data.timeout", "5s")
+ .set("spark.celeborn.push.replicate.enabled", enabled)
+ val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
+ val combineResult = combine(sparkSession)
+ val groupbyResult = groupBy(sparkSession)
+ val repartitionResult = repartition(sparkSession)
+ val sqlResult = runsql(sparkSession)
- Thread.sleep(3000L)
- sparkSession.stop()
+ Thread.sleep(3000L)
+ sparkSession.stop()
- val rssSparkSession = SparkSession.builder()
- .config(updateSparkConf(sparkConf, false)).getOrCreate()
- val rssCombineResult = combine(rssSparkSession)
- val rssGroupbyResult = groupBy(rssSparkSession)
- val rssRepartitionResult = repartition(rssSparkSession)
- val rssSqlResult = runsql(rssSparkSession)
+ val rssSparkSession = SparkSession.builder()
+ .config(updateSparkConf(sparkConf, false)).getOrCreate()
+ val rssCombineResult = combine(rssSparkSession)
+ val rssGroupbyResult = groupBy(rssSparkSession)
+ val rssRepartitionResult = repartition(rssSparkSession)
+ val rssSqlResult = runsql(rssSparkSession)
- assert(combineResult.equals(rssCombineResult))
- assert(groupbyResult.equals(rssGroupbyResult))
- assert(repartitionResult.equals(rssRepartitionResult))
- assert(combineResult.equals(rssCombineResult))
- assert(sqlResult.equals(rssSqlResult))
+ assert(combineResult.equals(rssCombineResult))
+ assert(groupbyResult.equals(rssGroupbyResult))
+ assert(repartitionResult.equals(rssRepartitionResult))
+ assert(combineResult.equals(rssCombineResult))
+ assert(sqlResult.equals(rssSqlResult))
- rssSparkSession.stop()
+ rssSparkSession.stop()
+ ShuffleClient.reset()
+ }
}
}
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
index 887b54b3..c3b999f5 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
@@ -132,8 +132,6 @@ class PushDataHandler extends BaseMessageHandler with Logging {
val mode = PartitionLocation.getMode(pushData.mode)
val body = pushData.body.asInstanceOf[NettyManagedBuffer].getBuf
val isMaster = mode == PartitionLocation.Mode.MASTER
- // Fetch real batchId from body will add more cost and no meaning for replicate.
- val batchId = if (isMaster) pushState.nextBatchId() else -1
// For test
if (isMaster && conf.testPushMasterDataTimeout && !pushMasterDataTimeoutTested) {
@@ -161,12 +159,17 @@ class PushDataHandler extends BaseMessageHandler with Logging {
partitionLocationInfo.getSlaveLocation(shuffleKey, pushData.partitionUniqueId)
}
+ // Fetch real batchId from body will add more cost and no meaning for replicate.
+ val doReplicate = location != null && location.getPeer != null && isMaster
+ val batchId = if (doReplicate) pushState.nextBatchId() else -1
val softSplit = new AtomicBoolean(false)
val wrappedCallback = new RpcResponseCallback() {
override def onSuccess(response: ByteBuffer): Unit = {
if (isMaster) {
- // Only master data will push data to slave
- pushState.removeBatch(batchId, location.hostAndPushPort())
+ if (doReplicate) {
+ // Only master data will push data to slave
+ pushState.removeBatch(batchId, location.hostAndPushPort())
+ }
workerSource.stopTimer(WorkerSource.MasterPushDataTime, key)
if (response.remaining() > 0) {
val resp = ByteBuffer.allocate(response.remaining())
@@ -187,7 +190,7 @@ class PushDataHandler extends BaseMessageHandler with Logging {
override def onFailure(e: Throwable): Unit = {
logError(s"[handlePushData.onFailure] partitionLocation: $location", e)
workerSource.incCounter(WorkerSource.PushDataFailCount)
- if (isMaster) {
+ if (doReplicate) {
pushState.removeBatch(batchId, location.hostAndPushPort())
}
// Throw by slave peer worker
@@ -285,7 +288,7 @@ class PushDataHandler extends BaseMessageHandler with Logging {
fileWriter.incrementPendingWrites()
// for master, send data to slave
- if (location.getPeer != null && isMaster) {
+ if (doReplicate) {
pushData.body().retain()
replicateThreadPool.submit(new Runnable {
override def run(): Unit = {
@@ -363,8 +366,6 @@ class PushDataHandler extends BaseMessageHandler with Logging {
val batchOffsets = pushMergedData.batchOffsets
val body = pushMergedData.body.asInstanceOf[NettyManagedBuffer].getBuf
val isMaster = mode == PartitionLocation.Mode.MASTER
- // Fetch real batchId from body will add more cost and no meaning for replicate.
- val batchId = if (isMaster) pushState.nextBatchId() else -1
val key = s"${pushMergedData.requestId}"
if (isMaster) {
@@ -392,11 +393,17 @@ class PushDataHandler extends BaseMessageHandler with Logging {
}
}
+ // Fetch real batchId from body will add more cost and no meaning for replicate.
+ val doReplicate = locations.head != null && locations.head.getPeer != null && isMaster
+ val batchId = if (doReplicate) pushState.nextBatchId() else -1
+
val wrappedCallback = new RpcResponseCallback() {
override def onSuccess(response: ByteBuffer): Unit = {
if (isMaster) {
- // Only master data will push data to slave
- pushState.removeBatch(batchId, locations.head.hostAndPushPort())
+ // Only master data enable replication will push data to slave
+ if (doReplicate) {
+ pushState.removeBatch(batchId, locations.head.hostAndPushPort())
+ }
workerSource.stopTimer(WorkerSource.MasterPushDataTime, key)
if (response.remaining() > 0) {
val resp = ByteBuffer.allocate(response.remaining())
@@ -414,7 +421,7 @@ class PushDataHandler extends BaseMessageHandler with Logging {
override def onFailure(e: Throwable): Unit = {
workerSource.incCounter(WorkerSource.PushDataFailCount)
- if (isMaster) {
+ if (doReplicate) {
pushState.removeBatch(batchId, locations.head.hostAndPushPort())
}
// Throw by slave peer worker
@@ -500,7 +507,7 @@ class PushDataHandler extends BaseMessageHandler with Logging {
fileWriters.foreach(_.incrementPendingWrites())
// for master, send data to slave
- if (locations.head.getPeer != null && isMaster) {
+ if (doReplicate) {
pushMergedData.body().retain()
replicateThreadPool.submit(new Runnable {
override def run(): Unit = {