This is an automated email from the ASF dual-hosted git repository.
ethanfeng 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 322f0d2b [CELEBORN-316] Wrap Celeborn exception with
CelebornIOException (#1253)
322f0d2b is described below
commit 322f0d2b41d415687aab26ddbb7c4ee29663bf51
Author: Angerszhuuuu <[email protected]>
AuthorDate: Wed Feb 22 16:10:11 2023 +0800
[CELEBORN-316] Wrap Celeborn exception with CelebornIOException (#1253)
---
.../apache/celeborn/client/ShuffleClientImpl.java | 31 +++++++++++-----------
.../celeborn/client/read/RssInputStream.java | 10 ++++---
.../client/read/WorkerPartitionReader.java | 9 ++++---
.../apache/celeborn/client/write/DataPusher.java | 17 ++++++------
.../celeborn/common/util/ExceptionUtils.java | 10 ++++---
.../common/write/InFlightRequestTracker.java | 5 ++--
.../common/write/SlowStartPushStrategy.java | 3 ++-
.../common/exception/CelebornIOException.scala} | 17 +++++-------
8 files changed, 54 insertions(+), 48 deletions(-)
diff --git
a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
index 238f19c1..adc604e4 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -39,6 +39,7 @@ import org.slf4j.LoggerFactory;
import org.apache.celeborn.client.compress.Compressor;
import org.apache.celeborn.client.read.RssInputStream;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.exception.CelebornIOException;
import org.apache.celeborn.common.haclient.RssHARetryClient;
import org.apache.celeborn.common.identity.UserIdentifier;
import org.apache.celeborn.common.network.TransportContext;
@@ -183,7 +184,7 @@ public class ShuffleClientImpl extends ShuffleClient {
if (!revive(
applicationId, shuffleId, mapId, attemptId, partitionId,
loc.getEpoch(), loc, cause)) {
callback.onFailure(
- new IOException("Revive Failed, remain revive times " +
remainReviveTimes));
+ new CelebornIOException("Revive Failed, remain revive times " +
remainReviveTimes));
} else if (mapperEnded(shuffleId, mapId, attemptId)) {
logger.debug(
"Retrying push data, but the mapper(map {} attempt {}) has ended.",
mapId, attemptId);
@@ -247,7 +248,7 @@ public class ShuffleClientImpl extends ShuffleClient {
} else {
pushState.exception.compareAndSet(
null,
- new IOException(
+ new CelebornIOException(
"Revive Failed in retry push merged data for location: " +
batch.loc));
return;
}
@@ -419,7 +420,7 @@ public class ShuffleClientImpl extends ShuffleClient {
boolean reachLimit = pushState.limitMaxInFlight(hostAndPushPort);
if (reachLimit) {
- throw new IOException("wait timeout for task " + mapKey,
pushState.exception.get());
+ throw new CelebornIOException("wait timeout for task " + mapKey,
pushState.exception.get());
}
}
@@ -427,7 +428,7 @@ public class ShuffleClientImpl extends ShuffleClient {
boolean reachLimit = pushState.limitZeroInFlight();
if (reachLimit) {
- throw new IOException("wait timeout for task " + mapKey,
pushState.exception.get());
+ throw new CelebornIOException("wait timeout for task " + mapKey,
pushState.exception.get());
}
}
@@ -552,7 +553,7 @@ public class ShuffleClientImpl extends ShuffleClient {
getPartitionLocation(applicationId, shuffleId, numMappers,
numPartitions);
if (map == null) {
- throw new IOException("Register shuffle failed for shuffle " +
shuffleKey);
+ throw new CelebornIOException("Register shuffle failed for shuffle " +
shuffleKey);
}
// get location
@@ -567,7 +568,7 @@ public class ShuffleClientImpl extends ShuffleClient {
-1,
null,
StatusCode.PUSH_DATA_FAIL_NON_CRITICAL_CAUSE)) {
- throw new IOException(
+ throw new CelebornIOException(
"Revive for shuffle " + shuffleKey + " partitionId " + partitionId
+ " failed.");
}
}
@@ -587,7 +588,7 @@ public class ShuffleClientImpl extends ShuffleClient {
final PartitionLocation loc = map.get(partitionId);
if (loc == null) {
- throw new IOException(
+ throw new CelebornIOException(
"Partition location for shuffle "
+ shuffleKey
+ " partitionId "
@@ -658,7 +659,7 @@ public class ShuffleClientImpl extends ShuffleClient {
@Override
public void onFailure(Throwable e) {
pushState.exception.compareAndSet(
- null, new IOException("Revived PushData failed!", e));
+ null, new CelebornIOException("Revived PushData failed!",
e));
logger.error(
"Push data to {}:{} failed for map {} attempt {} batch {}.",
loc.getHost(),
@@ -1025,7 +1026,7 @@ public class ShuffleClientImpl extends ShuffleClient {
+ " batches "
+ Arrays.toString(batchIds)
+ ".";
- pushState.exception.compareAndSet(null, new IOException(errorMsg,
e));
+ pushState.exception.compareAndSet(null, new
CelebornIOException(errorMsg, e));
if (logger.isDebugEnabled()) {
for (int batchId : batchIds) {
logger.debug(
@@ -1199,7 +1200,7 @@ public class ShuffleClientImpl extends ShuffleClient {
new MapperEnd(applicationId, shuffleId, mapId, attemptId,
numMappers, partitionId),
ClassTag$.MODULE$.apply(MapperEndResponse.class));
if (response.status() != StatusCode.SUCCESS) {
- throw new IOException("MapperEnd failed! StatusCode: " +
response.status());
+ throw new CelebornIOException("MapperEnd failed! StatusCode: " +
response.status());
}
} finally {
pushStates.remove(mapKey);
@@ -1211,7 +1212,7 @@ public class ShuffleClientImpl extends ShuffleClient {
final String mapKey = Utils.makeMapKey(shuffleId, mapId, attemptId);
PushState pushState = pushStates.remove(mapKey);
if (pushState != null) {
- pushState.exception.compareAndSet(null, new IOException("Cleaned Up"));
+ pushState.exception.compareAndSet(null, new CelebornIOException("Cleaned
Up"));
pushState.cleanup();
}
if (currentClient != null) {
@@ -1307,7 +1308,7 @@ public class ShuffleClientImpl extends ShuffleClient {
if (reduceFileGroups == null) {
String msg = "Shuffle data lost for shuffle " + shuffleId + " reduce " +
partitionId + "!";
logger.error(msg);
- throw new IOException(msg);
+ throw new CelebornIOException(msg);
}
return reduceFileGroups;
}
@@ -1507,7 +1508,7 @@ public class ShuffleClientImpl extends ShuffleClient {
}
if (!mapperEnded(shuffleId, mapId, attemptId)) {
pushState.exception.compareAndSet(
- null, new IOException("PushData byteBuf failed!", e));
+ null, new CelebornIOException("PushData byteBuf failed!",
e));
logger.error(
"Push data byteBuf to {}:{} failed for map {} attempt {}
batch {}.",
location.getHost(),
@@ -1670,7 +1671,7 @@ public class ShuffleClientImpl extends ShuffleClient {
shuffleId,
location.getId(),
location.getEpoch());
- throw new IOException("regiontstart revive failed");
+ throw new CelebornIOException("regiontstart revive failed");
}
}
return Optional.empty();
@@ -1779,7 +1780,7 @@ public class ShuffleClientImpl extends ShuffleClient {
if (currentException instanceof IOException) {
throw (IOException) currentException;
} else {
- throw new IOException(currentException.getMessage(), currentException);
+ throw new CelebornIOException(currentException.getMessage(),
currentException);
}
}
return result;
diff --git
a/client/src/main/java/org/apache/celeborn/client/read/RssInputStream.java
b/client/src/main/java/org/apache/celeborn/client/read/RssInputStream.java
index 66a18469..af8b4a8c 100644
--- a/client/src/main/java/org/apache/celeborn/client/read/RssInputStream.java
+++ b/client/src/main/java/org/apache/celeborn/client/read/RssInputStream.java
@@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
import org.apache.celeborn.client.compress.Decompressor;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.exception.CelebornIOException;
import org.apache.celeborn.common.network.client.TransportClientFactory;
import org.apache.celeborn.common.network.util.TransportConf;
import org.apache.celeborn.common.protocol.PartitionLocation;
@@ -242,7 +243,7 @@ public abstract class RssInputStream extends InputStream {
}
}
}
- throw new IOException("createPartitionReader failed!");
+ throw new CelebornIOException("createPartitionReader failed!");
}
private ByteBuf getNextChunk() throws IOException {
@@ -254,7 +255,8 @@ public abstract class RssInputStream extends InputStream {
currentReader.close();
if (fetchChunkRetryCnt == fetchChunkMaxRetry) {
logger.warn("Fetch chunk fail exceeds max retry {}",
fetchChunkRetryCnt);
- throw new IOException("Fetch chunk failed for " +
fetchChunkRetryCnt + " times");
+ throw new CelebornIOException(
+ "Fetch chunk failed for " + fetchChunkRetryCnt + " times");
} else {
if (currentReader.getLocation().getPeer() != null) {
logger.warn(
@@ -270,7 +272,7 @@ public abstract class RssInputStream extends InputStream {
}
}
}
- throw new IOException("Fetch chunk failed!");
+ throw new CelebornIOException("Fetch chunk failed!");
}
private PartitionReader createReader(
@@ -304,7 +306,7 @@ public abstract class RssInputStream extends InputStream {
conf, shuffleKey, location, clientFactory, startMapIndex,
endMapIndex);
}
- throw new IOException(
+ throw new CelebornIOException(
"Unknown storage info " + storageInfo + " to read location " +
location);
}
diff --git
a/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
b/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
index 2a9bae7b..2058fa8b 100644
---
a/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
+++
b/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
@@ -29,6 +29,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.exception.CelebornIOException;
import org.apache.celeborn.common.network.buffer.ManagedBuffer;
import org.apache.celeborn.common.network.buffer.NettyManagedBuffer;
import org.apache.celeborn.common.network.client.ChunkReceivedCallback;
@@ -92,14 +93,14 @@ public class WorkerPartitionReader implements
PartitionReader {
public void onFailure(int chunkIndex, Throwable e) {
String errorMsg = "Fetch chunk " + chunkIndex + " failed.";
logger.error(errorMsg, e);
- exception.set(new IOException(errorMsg, e));
+ exception.set(new CelebornIOException(errorMsg, e));
}
};
TransportClient client;
try {
client = clientFactory.createClient(location.getHost(),
location.getFetchPort());
} catch (InterruptedException ie) {
- throw new IOException("Interrupted when createClient", ie);
+ throw new CelebornIOException("Interrupted when createClient", ie);
}
OpenStream openBlocks =
new OpenStream(shuffleKey, location.getFileName(), startMapIndex,
endMapIndex);
@@ -131,7 +132,7 @@ public class WorkerPartitionReader implements
PartitionReader {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- IOException ioe = new IOException(e);
+ IOException ioe = new CelebornIOException(e);
exception.set(ioe);
throw ioe;
}
@@ -161,7 +162,7 @@ public class WorkerPartitionReader implements
PartitionReader {
Math.min(fetchMaxReqsInFlight - inFlight + 1, streamHandle.numChunks
- chunkIndex);
for (int i = 0; i < toFetch; i++) {
if (testFetch && fetchChunkRetryCnt < fetchChunkMaxRetry - 1 &&
chunkIndex == 3) {
- callback.onFailure(chunkIndex, new IOException("Test fetch chunk
failure"));
+ callback.onFailure(chunkIndex, new CelebornIOException("Test fetch
chunk failure"));
} else {
try {
TransportClient client =
diff --git
a/client/src/main/java/org/apache/celeborn/client/write/DataPusher.java
b/client/src/main/java/org/apache/celeborn/client/write/DataPusher.java
index ccb586dd..7ad87791 100644
--- a/client/src/main/java/org/apache/celeborn/client/write/DataPusher.java
+++ b/client/src/main/java/org/apache/celeborn/client/write/DataPusher.java
@@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory;
import org.apache.celeborn.client.ShuffleClient;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.exception.CelebornIOException;
public class DataPusher {
private static final Logger logger =
LoggerFactory.getLogger(DataPusher.class);
@@ -85,7 +86,7 @@ public class DataPusher {
idleQueue.put(new PushTask(pushBufferMaxSize));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- throw new IOException(e);
+ throw new CelebornIOException(e);
}
}
@@ -110,7 +111,7 @@ public class DataPusher {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- exceptionRef.set(new IOException(e));
+ exceptionRef.set(new CelebornIOException(e));
} finally {
idleLock.unlock();
}
@@ -126,10 +127,8 @@ public class DataPusher {
}
pushData(task);
reclaimTask(task);
- } catch (InterruptedException e) {
- exceptionRef.set(new IOException(e));
- } catch (IOException e) {
- exceptionRef.set(e);
+ } catch (InterruptedException | IOException e) {
+ exceptionRef.set(new CelebornIOException(e));
}
}
}
@@ -153,7 +152,7 @@ public class DataPusher {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- IOException ioe = new IOException(e);
+ IOException ioe = new CelebornIOException(e);
exceptionRef.set(ioe);
throw ioe;
}
@@ -165,7 +164,7 @@ public class DataPusher {
waitIdleQueueFullWithLock();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- exceptionRef.set(new IOException(e));
+ exceptionRef.set(new CelebornIOException(e));
}
terminated = true;
@@ -209,7 +208,7 @@ public class DataPusher {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- exceptionRef.set(new IOException(e));
+ exceptionRef.set(new CelebornIOException(e));
} finally {
idleLock.unlock();
}
diff --git
a/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
b/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
index 5336f6d1..119351d2 100644
--- a/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
+++ b/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
@@ -19,13 +19,17 @@ package org.apache.celeborn.common.util;
import java.io.IOException;
+import org.apache.celeborn.common.exception.CelebornIOException;
+
public class ExceptionUtils {
public static void wrapAndThrowIOException(Exception exception) throws
IOException {
- if (exception instanceof IOException) {
- throw (IOException) exception;
+ if (exception instanceof CelebornIOException) {
+ throw (CelebornIOException) exception;
+ } else if (exception instanceof IOException) {
+ throw new CelebornIOException(exception);
} else {
- throw new IOException(exception.getMessage(), exception);
+ throw new CelebornIOException(exception.getMessage(), exception);
}
}
}
diff --git
a/common/src/main/java/org/apache/celeborn/common/write/InFlightRequestTracker.java
b/common/src/main/java/org/apache/celeborn/common/write/InFlightRequestTracker.java
index 3b3e8da6..c311d910 100644
---
a/common/src/main/java/org/apache/celeborn/common/write/InFlightRequestTracker.java
+++
b/common/src/main/java/org/apache/celeborn/common/write/InFlightRequestTracker.java
@@ -27,6 +27,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.exception.CelebornIOException;
/*
* This class is for track in flight request and limit request.
@@ -97,7 +98,7 @@ public class InFlightRequestTracker {
times--;
}
} catch (InterruptedException e) {
- pushState.exception.set(new IOException(e));
+ pushState.exception.set(new CelebornIOException(e));
}
if (times <= 0) {
@@ -141,7 +142,7 @@ public class InFlightRequestTracker {
times--;
}
} catch (InterruptedException e) {
- pushState.exception.set(new IOException(e));
+ pushState.exception.set(new CelebornIOException(e));
}
if (times <= 0) {
diff --git
a/common/src/main/java/org/apache/celeborn/common/write/SlowStartPushStrategy.java
b/common/src/main/java/org/apache/celeborn/common/write/SlowStartPushStrategy.java
index 5d29b764..2517507d 100644
---
a/common/src/main/java/org/apache/celeborn/common/write/SlowStartPushStrategy.java
+++
b/common/src/main/java/org/apache/celeborn/common/write/SlowStartPushStrategy.java
@@ -26,6 +26,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.exception.CelebornIOException;
public class SlowStartPushStrategy extends PushStrategy {
@@ -153,7 +154,7 @@ public class SlowStartPushStrategy extends PushStrategy {
logger.debug("Will sleep {} ms to control the push speed.",
sleepInterval);
Thread.sleep(sleepInterval);
} catch (InterruptedException e) {
- pushState.exception.set(new IOException(e));
+ pushState.exception.set(new CelebornIOException(e));
}
}
}
diff --git
a/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
b/common/src/main/scala/org/apache/celeborn/common/exception/CelebornIOException.scala
similarity index 69%
copy from
common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
copy to
common/src/main/scala/org/apache/celeborn/common/exception/CelebornIOException.scala
index 5336f6d1..7f11d3ab 100644
--- a/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
+++
b/common/src/main/scala/org/apache/celeborn/common/exception/CelebornIOException.scala
@@ -15,17 +15,14 @@
* limitations under the License.
*/
-package org.apache.celeborn.common.util;
+package org.apache.celeborn.common.exception
-import java.io.IOException;
+import java.io.IOException
-public class ExceptionUtils {
+class CelebornIOException(message: String, cause: Throwable)
+ extends IOException(message, cause) {
- public static void wrapAndThrowIOException(Exception exception) throws
IOException {
- if (exception instanceof IOException) {
- throw (IOException) exception;
- } else {
- throw new IOException(exception.getMessage(), exception);
- }
- }
+ def this(message: String) = this(message, null)
+
+ def this(cause: Throwable) = this(cause.getMessage, cause)
}
