[SPARK-34722][CORE][SQL][TEST] Clean up deprecated API usage related to JUnit4

### What changes were proposed in this pull request?
The main change of this pr as follows:

- Use `org.junit.Assert.assertThrows(String, Class, ThrowingRunnable)` method instead of  `ExpectedException.none()`
- Use `org.hamcrest.MatcherAssert.assertThat()` method instead of   `org.junit.Assert.assertThat(T, org.hamcrest.Matcher<? super T>)`

### Why are the changes needed?
Clean up deprecated API usage

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Pass the Jenkins or GitHub Action

Closes #31815 from LuciferYang/SPARK-34722.

Authored-by: yangjie01 <yangjie01@baidu.com>
Signed-off-by: Dongjoon Hyun <dhyun@apple.com>

common/network-common/src/test/java/org/apache/spark/network/client/TransportClientFactorySuite.java

@@ -29,9 +29,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.ExpectedException;
 
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotSame;
@@ -227,11 +225,8 @@ public class TransportClientFactorySuite {
     factory.createClient(TestUtils.getLocalHost(), server1.getPort());
   }
 
-  @Rule
-  public ExpectedException expectedException = ExpectedException.none();
-
   @Test
-  public void fastFailConnectionInTimeWindow() throws IOException, InterruptedException {
+  public void fastFailConnectionInTimeWindow() {
     TransportClientFactory factory = context.createClientFactory();
     TransportServer server = context.createServer();
     int unreachablePort = server.getPort();
@@ -241,9 +236,7 @@ public class TransportClientFactorySuite {
     } catch (Exception e) {
       assert(e instanceof IOException);
     }
-    expectedException.expect(IOException.class);
-    expectedException.expectMessage("fail this connection directly");
-    factory.createClient(TestUtils.getLocalHost(), unreachablePort, true);
-    expectedException = ExpectedException.none();
+    Assert.assertThrows("fail this connection directly", IOException.class,
+      () -> factory.createClient(TestUtils.getLocalHost(), unreachablePort, true));
   }
 }

common/network-common/src/test/java/org/apache/spark/network/crypto/TransportCipherSuite.java

@@ -29,11 +29,11 @@ import org.apache.commons.crypto.stream.CryptoOutputStream;
 import org.apache.spark.network.util.MapConfigProvider;
 import org.apache.spark.network.util.TransportConf;
 import org.hamcrest.CoreMatchers;
+import org.hamcrest.MatcherAssert;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
@@ -81,7 +81,7 @@ public class TransportCipherSuite {
       channel.writeInbound(buffer2);
       fail("Should have raised an exception");
     } catch (Throwable expected) {
-      assertThat(expected, CoreMatchers.instanceOf(IOException.class));
+      MatcherAssert.assertThat(expected, CoreMatchers.instanceOf(IOException.class));
       assertEquals(0, buffer2.refCnt());
     }
 

core/src/test/java/org/apache/spark/unsafe/map/AbstractBytesToBytesMapSuite.java

@@ -24,6 +24,7 @@ import java.util.*;
 
 import scala.Tuple2$;
 
+import org.hamcrest.MatcherAssert;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -539,7 +540,7 @@ public abstract class AbstractBytesToBytesMapSuite {
           break;
         }
       }
-      Assert.assertThat(i, greaterThan(0));
+      MatcherAssert.assertThat(i, greaterThan(0));
       Assert.assertFalse(success);
     } finally {
       map.free();

core/src/test/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorterSuite.java

@@ -25,6 +25,7 @@ import java.util.UUID;
 
 import scala.Tuple2$;
 
+import org.hamcrest.MatcherAssert;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -224,7 +225,7 @@ public class UnsafeExternalSorterSuite {
 
     sorter.insertRecord(null, 0, 0, 0, false);
     sorter.spill();
-    assertThat(sorter.getSortTimeNanos(), greaterThan(prevSortTime));
+    MatcherAssert.assertThat(sorter.getSortTimeNanos(), greaterThan(prevSortTime));
     prevSortTime = sorter.getSortTimeNanos();
 
     sorter.spill();  // no sort needed
@@ -232,7 +233,7 @@ public class UnsafeExternalSorterSuite {
 
     sorter.insertRecord(null, 0, 0, 0, false);
     UnsafeSorterIterator iter = sorter.getSortedIterator();
-    assertThat(sorter.getSortTimeNanos(), greaterThan(prevSortTime));
+    MatcherAssert.assertThat(sorter.getSortTimeNanos(), greaterThan(prevSortTime));
 
     sorter.cleanupResources();
     assertSpillFilesWereCleanedUp();
@@ -251,7 +252,7 @@ public class UnsafeExternalSorterSuite {
     // The insertion of this record should trigger a spill:
     insertNumber(sorter, 0);
     // Ensure that spill files were created
-    assertThat(tempDir.listFiles().length, greaterThanOrEqualTo(1));
+    MatcherAssert.assertThat(tempDir.listFiles().length, greaterThanOrEqualTo(1));
     // Read back the sorted data:
     UnsafeSorterIterator iter = sorter.getSortedIterator();
 

launcher/src/test/java/org/apache/spark/launcher/SparkSubmitCommandBuilderSuite.java

@@ -26,11 +26,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.regex.Pattern;
 
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
+import org.junit.*;
 
 import static org.junit.Assert.*;
 
@@ -39,9 +35,6 @@ public class SparkSubmitCommandBuilderSuite extends BaseSuite {
   private static File dummyPropsFile;
   private static SparkSubmitOptionParser parser;
 
-  @Rule
-  public ExpectedException expectedException = ExpectedException.none();
-
   @BeforeClass
   public static void setUp() throws Exception {
     dummyPropsFile = File.createTempFile("spark", "properties");
@@ -216,15 +209,13 @@ public class SparkSubmitCommandBuilderSuite extends BaseSuite {
 
   @Test
   public void testExamplesRunnerWithMasterNoMainClass() throws Exception {
-    expectedException.expect(IllegalArgumentException.class);
-    expectedException.expectMessage("Missing example class name.");
-
     List<String> sparkSubmitArgs = Arrays.asList(
       SparkSubmitCommandBuilder.RUN_EXAMPLE,
       parser.MASTER + "=foo"
     );
     Map<String, String> env = new HashMap<>();
-    buildCommand(sparkSubmitArgs, env);
+    Assert.assertThrows("Missing example class name.", IllegalArgumentException.class,
+      () -> buildCommand(sparkSubmitArgs, env));
   }
 
   @Test

sql/core/src/test/java/test/org/apache/spark/sql/JavaDatasetSuite.java

@@ -37,7 +37,6 @@ import scala.Tuple5;
 
 import com.google.common.base.Objects;
 import org.junit.*;
-import org.junit.rules.ExpectedException;
 
 import org.apache.spark.api.java.JavaPairRDD;
 import org.apache.spark.api.java.JavaSparkContext;
@@ -911,9 +910,6 @@ public class JavaDatasetSuite implements Serializable {
     }
   }
 
-  @Rule
-  public transient ExpectedException nullabilityCheck = ExpectedException.none();
-
   @Test
   public void testRuntimeNullabilityCheck() {
     OuterScopes.addOuterScope(this);
@@ -955,9 +951,6 @@ public class JavaDatasetSuite implements Serializable {
       Assert.assertEquals(Collections.singletonList(nestedSmallBean), ds.collectAsList());
     }
 
-    nullabilityCheck.expect(RuntimeException.class);
-    nullabilityCheck.expectMessage("Null value appeared in non-nullable field");
-
     {
       Row row = new GenericRow(new Object[] {
           new GenericRow(new Object[] {
@@ -968,7 +961,8 @@ public class JavaDatasetSuite implements Serializable {
       Dataset<Row> df = spark.createDataFrame(Collections.singletonList(row), schema);
       Dataset<NestedSmallBean> ds = df.as(Encoders.bean(NestedSmallBean.class));
 
-      ds.collect();
+      Assert.assertThrows("Null value appeared in non-nullable field", RuntimeException.class,
+        ds::collect);
     }
   }