[SPARK-38613][CORE] Change the exception type thrown by PushBlockStreamCallback#abortIfNecessary from RuntimeException to IllegalStateException

[LuciferYang]

What changes were proposed in this pull request?

This pr change the exception type thrown by PushBlockStreamCallback#abortIfNecessary from RuntimeException to IllegalStateException and fixed the corresponding test case.

In addition, this PR fixes the bug of testWritingPendingBufsIsAbortedImmediatelyDuringComplete in RemoteBlockPushResolverSuite. RuntimeException throw by

spark/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RemoteBlockPushResolverSuite.java

Lines 818 to 820 in acb50d9

	RemoteBlockPushResolver.PushBlockStreamCallback callback2 =
	(RemoteBlockPushResolver.PushBlockStreamCallback) pushResolver.receiveBlockDataAsStream(
	new PushBlockStream(TEST_APP, 1, 0, 0, 5, 0, 0));

before this pr, but the test case expects it to be thrown by

spark/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RemoteBlockPushResolverSuite.java

Lines 834 to 840 in 2ca5d18

	try {
	callback.onComplete(callback.getID());
	} catch (Throwable t) {
	assertEquals("IOExceptions exceeded the threshold when merging shufflePush_0_0_0",
	t.getMessage());
	throw t;
	}

Why are the changes needed?

The RuntimeException is too broad, it should be specific.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Pass GA and fix related UTs

[LuciferYang]

cc @otterc, also ping @dongjoon-hyun @HyukjinKwon @srowen

[LuciferYang]

On the other hand, should we uniformly use Assert.assertThrows to replace @Test(expected = Exception.class)? My idea is:

Assert.assertThrows can accurately ensure a statement will throw an Exception, but @Test(expected = Exception.class) only ensure the current Test method will throw the corresponding Exception.

@srowen @dongjoon-hyun @HyukjinKwon If necessary, I will file a new Jira.

[srowen]

I wouldn't change everything, but change cases where we suspect that it's catching too much.

The problem here is really looking for RuntimeException, not the scope. That is too broad. Is something more specific thrown that we can catch without catching other stuff?

[LuciferYang]

The exception thrown from

spark/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

Lines 761 to 768 in acb50d9

	private void abortIfNecessary() {
	if (partitionInfo.shouldAbort(mergeManager.ioExceptionsThresholdDuringMerge)) {
	deferredBufs = null;
	throw new RuntimeException(String.format("%s when merging %s",
	ErrorHandler.BlockPushErrorHandler.IOEXCEPTIONS_EXCEEDED_THRESHOLD_PREFIX,
	streamId));
	}
	}

Only RuntimeException is thrown here now, no more specific type. BlockPushNonFatalFailure is not suitable here, may be necessary to define a new type.

[srowen]

We could fix that. This should really be something more specific - IllegalStateException? this is just the problem it causes, using RuntimeException

[LuciferYang]

We could fix that. This should really be something more specific - IllegalStateException? this is just the problem it causes, using RuntimeException

ok

[LuciferYang]

@srowen b08a255 change exception type from RuntimeException to IllegalStateException

[srowen]

No big deal but this can be * @throws IllegalStateException if ...

[LuciferYang]

4583b22 fix this

[srowen]

Looks OK pending tests

[srowen]

Merged to master

[LuciferYang]

thanks @srowen