Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/18472#discussion_r124954827
--- Diff:
common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java
---
@@ -104,15 +106,31 @@ public void deactivateStream() {
*/
private void failOutstandingRequests(Throwable cause) {
for (Map.Entry<StreamChunkId, ChunkReceivedCallback> entry :
outstandingFetches.entrySet()) {
- entry.getValue().onFailure(entry.getKey().chunkIndex, cause);
+ try {
+ entry.getValue().onFailure(entry.getKey().chunkIndex, cause);
+ } catch (Exception e) {
+ logger.warn("ChunkReceivedCallback.onFailure throws exception", e);
+ }
}
for (Map.Entry<Long, RpcResponseCallback> entry :
outstandingRpcs.entrySet()) {
- entry.getValue().onFailure(cause);
+ try {
+ entry.getValue().onFailure(cause);
+ } catch (Exception e) {
+ logger.warn("RpcResponseCallback.onFailure throws exception", e);
+ }
+ }
+ for (Tuple2<String, StreamCallback> entry : streamCallbacks) {
+ try {
+ entry._2().onFailure(entry._1(), cause);
--- End diff --
Actually `StreamCallback` has a 1-1 mapping to the steam, so ideally we
don't need to pass `streamId` when calling `onFailure` or `onSuccess`, the
place that creates the `StreamCallback` already knows which stream this
callback is resposible to. I also checked all implemetations of
`StreamCallback`, all of them ignore the `streamId` parameter. Shall we update
the `StreamCallback` interface?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
