diff --git a/.baseline/checkstyle/checkstyle.xml b/.baseline/checkstyle/checkstyle.xml
index 9b22c83c24ae..278b385b6b77 100644
--- a/.baseline/checkstyle/checkstyle.xml
+++ b/.baseline/checkstyle/checkstyle.xml
@@ -410,6 +410,11 @@
             <property name="format" value="(void setUp\(\))|(void setup\(\))|(void setupStatic\(\))|(void setUpStatic\(\))|(void beforeTest\(\))|(void teardown\(\))|(void tearDown\(\))|(void beforeStatic\(\))|(void afterStatic\(\))"/>
             <property name="message" value="Test setup/teardown methods are called before(), beforeClass(), after(), afterClass(), but not setUp, teardown, etc."/>
         </module>
+        <module name="RegexpSinglelineJava">
+            <property name="ignoreComments" value="true"/>
+            <property name="format" value="@Test\(.*expected.*\)"/>
+            <property name="message" value="Prefer using Assertions.assertThatThrownBy(...).isInstanceOf(...) instead."/>
+        </module>
         <module name="RightCurly"> <!-- Java Style Guide: Nonempty blocks: K & R style -->
             <property name="option" value="same"/>
             <property name="tokens" value="LITERAL_TRY, LITERAL_CATCH, LITERAL_FINALLY, LITERAL_IF, LITERAL_ELSE, LITERAL_DO"/>
diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java
index d55057e9be87..50ed13dceee4 100644
--- a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java
+++ b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java
@@ -51,19 +51,17 @@ public class TestExpressionBinding {
   @Test
   public void testMissingReference() {
     Expression expr = and(equal("t", 5), equal("x", 7));
-    try {
-      Binder.bind(STRUCT, expr);
-      Assert.fail("Should not successfully bind to struct without field 't'");
-    } catch (ValidationException e) {
-      Assert.assertTrue("Should complain about missing field",
-          e.getMessage().contains("Cannot find field 't' in struct:"));
-    }
+    Assertions.assertThatThrownBy(() -> Binder.bind(STRUCT, expr))
+        .isInstanceOf(ValidationException.class)
+        .hasMessageContaining("Cannot find field 't' in struct");
   }
 
-  @Test(expected = IllegalStateException.class)
+  @Test
   public void testBoundExpressionFails() {
     Expression expr = not(equal("x", 7));
-    Binder.bind(STRUCT, Binder.bind(STRUCT, expr));
+    Assertions.assertThatThrownBy(() -> Binder.bind(STRUCT, Binder.bind(STRUCT, expr)))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("Found already bound predicate");
   }
 
   @Test
@@ -78,10 +76,12 @@ public void testCaseInsensitiveReference() {
     TestHelpers.assertAllReferencesBound("Single reference", Binder.bind(STRUCT, expr, false));
   }
 
-  @Test(expected = ValidationException.class)
+  @Test
   public void testCaseSensitiveReference() {
     Expression expr = not(equal("X", 7));
-    Binder.bind(STRUCT, expr, true);
+    Assertions.assertThatThrownBy(() -> Binder.bind(STRUCT, expr, true))
+        .isInstanceOf(ValidationException.class)
+        .hasMessageContaining("Cannot find field 'X' in struct");
   }
 
   @Test
diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestStringLiteralConversions.java b/api/src/test/java/org/apache/iceberg/expressions/TestStringLiteralConversions.java
index d2812a731817..afc725c8b2c3 100644
--- a/api/src/test/java/org/apache/iceberg/expressions/TestStringLiteralConversions.java
+++ b/api/src/test/java/org/apache/iceberg/expressions/TestStringLiteralConversions.java
@@ -31,6 +31,7 @@
 import org.apache.avro.Schema;
 import org.apache.avro.data.TimeConversions;
 import org.apache.iceberg.types.Types;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -159,18 +160,22 @@ public void testNegativeStringToTimestampLiteral() {
 
   }
 
-  @Test(expected = DateTimeException.class)
+  @Test
   public void testTimestampWithZoneWithoutZoneInLiteral() {
     // Zone must be present in literals when converting to timestamp with zone
     Literal<CharSequence> timestampStr = Literal.of("2017-08-18T14:21:01.919");
-    timestampStr.to(Types.TimestampType.withZone());
+    Assertions.assertThatThrownBy(() -> timestampStr.to(Types.TimestampType.withZone()))
+        .isInstanceOf(DateTimeException.class)
+        .hasMessageContaining("could not be parsed");
   }
 
-  @Test(expected = DateTimeException.class)
+  @Test
   public void testTimestampWithoutZoneWithZoneInLiteral() {
     // Zone must not be present in literals when converting to timestamp without zone
     Literal<CharSequence> timestampStr = Literal.of("2017-08-18T14:21:01.919+07:00");
-    timestampStr.to(Types.TimestampType.withoutZone());
+    Assertions.assertThatThrownBy(() -> timestampStr.to(Types.TimestampType.withoutZone()))
+        .isInstanceOf(DateTimeException.class)
+        .hasMessageContaining("could not be parsed");
   }
 
   @Test
diff --git a/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java b/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
index 8257b9e50ed4..9560009f76cc 100644
--- a/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
+++ b/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
@@ -30,6 +30,7 @@
 import org.apache.iceberg.expressions.ResidualEvaluator;
 import org.apache.iceberg.expressions.UnboundPredicate;
 import org.apache.iceberg.types.Types;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -129,7 +130,7 @@ public void testCaseInsensitiveIdentityTransformResiduals() {
     Assert.assertEquals("Residual should be alwaysFalse", alwaysFalse(), residual);
   }
 
-  @Test(expected = ValidationException.class)
+  @Test
   public void testCaseSensitiveIdentityTransformResiduals() {
     Schema schema = new Schema(
         Types.NestedField.optional(50, "dateint", Types.IntegerType.get()),
@@ -142,7 +143,9 @@ public void testCaseSensitiveIdentityTransformResiduals() {
 
     ResidualEvaluator resEval = ResidualEvaluator.of(spec, lessThan("DATEINT", 20170815), true);
 
-    resEval.residualFor(Row.of(20170815));
+    Assertions.assertThatThrownBy(() -> resEval.residualFor(Row.of(20170815)))
+        .isInstanceOf(ValidationException.class)
+        .hasMessageContaining("Cannot find field 'DATEINT' in struct");
   }
 
   @Test
diff --git a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
index 210efd352f5b..5e64f8a712ce 100644
--- a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
+++ b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
@@ -26,6 +26,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Types.IntegerType;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -427,7 +428,7 @@ public void testProjectMapNested() {
     Assert.assertEquals(expected.asStruct(), actual.asStruct());
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testReassignIdsIllegalArgumentException() {
     Schema schema = new Schema(
         required(1, "a", Types.IntegerType.get()),
@@ -436,10 +437,12 @@ public void testReassignIdsIllegalArgumentException() {
     Schema sourceSchema = new Schema(
         required(1, "a", Types.IntegerType.get())
     );
-    TypeUtil.reassignIds(schema, sourceSchema);
+    Assertions.assertThatThrownBy(() -> TypeUtil.reassignIds(schema, sourceSchema))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Field b not found in source schema");
   }
 
-  @Test(expected = RuntimeException.class)
+  @Test
   public void testValidateSchemaViaIndexByName() {
     Types.NestedField nestedType = Types.NestedField
         .required(1, "a", Types.StructType.of(
@@ -450,7 +453,9 @@ public void testValidateSchemaViaIndexByName() {
             )
         );
 
-    TypeUtil.indexByName(Types.StructType.of(nestedType));
+    Assertions.assertThatThrownBy(() -> TypeUtil.indexByName(Types.StructType.of(nestedType)))
+            .isInstanceOf(RuntimeException.class)
+            .hasMessageContaining("Invalid schema: multiple fields for name a.b.c");
   }
 
   @Test
diff --git a/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/parquet/DecimalVectorUtilTest.java b/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/parquet/DecimalVectorUtilTest.java
index 10fe1afcdfbf..e62c6ae7ba2a 100644
--- a/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/parquet/DecimalVectorUtilTest.java
+++ b/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/parquet/DecimalVectorUtilTest.java
@@ -20,6 +20,7 @@
 package org.apache.iceberg.arrow.vectorized.parquet;
 
 import java.math.BigInteger;
+import org.assertj.core.api.Assertions;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
@@ -65,9 +66,11 @@ public void testPadBigEndianBytesZero() {
     assertEquals(BigInteger.ZERO, result);
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testPadBigEndianBytesOverflow() {
     byte[] bytes = new byte[17];
-    DecimalVectorUtil.padBigEndianBytes(bytes, 16);
+    Assertions.assertThatThrownBy(() -> DecimalVectorUtil.padBigEndianBytes(bytes, 16))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Buffer size of 17 is larger than requested size of 16");
   }
 }
diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopMetricsContext.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopMetricsContext.java
index 359122608ec2..adff76c25095 100644
--- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopMetricsContext.java
+++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopMetricsContext.java
@@ -51,8 +51,8 @@ public void initialize(Map<String, String> properties) {
   }
 
   /**
-   * The Hadoop implementation delegates to the Hadoop delegates to the
-   * FileSystem.Statistics implementation and therefore does not require
+   * The Hadoop implementation delegates to the FileSystem.Statistics
+   * implementation and therefore does not require
    * support for operations like unit() and count() as the counter
    * values are not directly consumed.
    *
diff --git a/data/src/test/java/org/apache/iceberg/data/avro/TestSingleMessageEncoding.java b/data/src/test/java/org/apache/iceberg/data/avro/TestSingleMessageEncoding.java
index 15c44761981e..84ac6062063b 100644
--- a/data/src/test/java/org/apache/iceberg/data/avro/TestSingleMessageEncoding.java
+++ b/data/src/test/java/org/apache/iceberg/data/avro/TestSingleMessageEncoding.java
@@ -38,6 +38,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Ordering;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Types;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -132,14 +133,16 @@ public void testSchemaEvolution() throws Exception {
     Assert.assertEquals(allAsV2, decodedUsingV2);
   }
 
-  @Test(expected = MissingSchemaException.class)
+  @Test
   public void testCompatibleReadFailsWithoutSchema() throws Exception {
     MessageEncoder<Record> v1Encoder = new IcebergEncoder<>(SCHEMA_V1);
     MessageDecoder<Record> v2Decoder = new IcebergDecoder<>(SCHEMA_V2);
 
     ByteBuffer v1Buffer = v1Encoder.encode(V1_RECORDS.get(3));
 
-    v2Decoder.decode(v1Buffer);
+    Assertions.assertThatThrownBy(() -> v2Decoder.decode(v1Buffer))
+            .isInstanceOf(MissingSchemaException.class)
+            .hasMessageContaining("Cannot resolve schema for fingerprint");
   }
 
   @Test
@@ -202,7 +205,7 @@ public void testBufferCopy() throws Exception {
         V1_RECORDS.get(0), decoder.decode(b0));
   }
 
-  @Test(expected = AvroRuntimeException.class)
+  @Test
   public void testByteBufferMissingPayload() throws Exception {
     MessageEncoder<Record> encoder = new IcebergEncoder<>(SCHEMA_V2);
     MessageDecoder<Record> decoder = new IcebergDecoder<>(SCHEMA_V2);
@@ -211,10 +214,12 @@ public void testByteBufferMissingPayload() throws Exception {
 
     buffer.limit(12);
 
-    decoder.decode(buffer);
+    Assertions.assertThatThrownBy(() -> decoder.decode(buffer))
+            .isInstanceOf(AvroRuntimeException.class)
+            .hasMessage("Decoding datum failed");
   }
 
-  @Test(expected = BadHeaderException.class)
+  @Test
   public void testByteBufferMissingFullHeader() throws Exception {
     MessageEncoder<Record> encoder = new IcebergEncoder<>(SCHEMA_V2);
     MessageDecoder<Record> decoder = new IcebergDecoder<>(SCHEMA_V2);
@@ -223,10 +228,12 @@ public void testByteBufferMissingFullHeader() throws Exception {
 
     buffer.limit(8);
 
-    decoder.decode(buffer);
+    Assertions.assertThatThrownBy(() -> decoder.decode(buffer))
+            .isInstanceOf(BadHeaderException.class)
+            .hasMessage("Not enough header bytes");
   }
 
-  @Test(expected = BadHeaderException.class)
+  @Test
   public void testByteBufferBadMarkerByte() throws Exception {
     MessageEncoder<Record> encoder = new IcebergEncoder<>(SCHEMA_V2);
     MessageDecoder<Record> decoder = new IcebergDecoder<>(SCHEMA_V2);
@@ -234,10 +241,12 @@ public void testByteBufferBadMarkerByte() throws Exception {
     ByteBuffer buffer = encoder.encode(V2_RECORDS.get(0));
     buffer.array()[0] = 0x00;
 
-    decoder.decode(buffer);
+    Assertions.assertThatThrownBy(() -> decoder.decode(buffer))
+            .isInstanceOf(BadHeaderException.class)
+            .hasMessageContaining("Unrecognized header bytes");
   }
 
-  @Test(expected = BadHeaderException.class)
+  @Test
   public void testByteBufferBadVersionByte() throws Exception {
     MessageEncoder<Record> encoder = new IcebergEncoder<>(SCHEMA_V2);
     MessageDecoder<Record> decoder = new IcebergDecoder<>(SCHEMA_V2);
@@ -245,10 +254,12 @@ public void testByteBufferBadVersionByte() throws Exception {
     ByteBuffer buffer = encoder.encode(V2_RECORDS.get(0));
     buffer.array()[1] = 0x00;
 
-    decoder.decode(buffer);
+    Assertions.assertThatThrownBy(() -> decoder.decode(buffer))
+            .isInstanceOf(BadHeaderException.class)
+            .hasMessageContaining("Unrecognized header bytes");
   }
 
-  @Test(expected = MissingSchemaException.class)
+  @Test
   public void testByteBufferUnknownSchema() throws Exception {
     MessageEncoder<Record> encoder = new IcebergEncoder<>(SCHEMA_V2);
     MessageDecoder<Record> decoder = new IcebergDecoder<>(SCHEMA_V2);
@@ -256,6 +267,8 @@ public void testByteBufferUnknownSchema() throws Exception {
     ByteBuffer buffer = encoder.encode(V2_RECORDS.get(0));
     buffer.array()[4] = 0x00;
 
-    decoder.decode(buffer);
+    Assertions.assertThatThrownBy(() -> decoder.decode(buffer))
+            .isInstanceOf(MissingSchemaException.class)
+            .hasMessageContaining("Cannot resolve schema for fingerprint");
   }
 }
diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/HiveTableTest.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/HiveTableTest.java
index 605621309092..9bfc969ac4c4 100644
--- a/hive-metastore/src/test/java/org/apache/iceberg/hive/HiveTableTest.java
+++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/HiveTableTest.java
@@ -65,6 +65,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.types.Types;
 import org.apache.thrift.TException;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
@@ -285,16 +286,18 @@ public void testColumnTypeChangeInMetastore() throws TException {
     Assert.assertEquals("Schema should match expected", expectedSchema.asStruct(), icebergTable.schema().asStruct());
   }
 
-  @Test(expected = CommitFailedException.class)
+  @Test
   public void testFailure() throws TException {
     Table icebergTable = catalog.loadTable(TABLE_IDENTIFIER);
     org.apache.hadoop.hive.metastore.api.Table table = metastoreClient.getTable(DB_NAME, TABLE_NAME);
     String dummyLocation = "dummylocation";
     table.getParameters().put(METADATA_LOCATION_PROP, dummyLocation);
     metastoreClient.alter_table(DB_NAME, TABLE_NAME, table);
-    icebergTable.updateSchema()
-        .addColumn("data", Types.LongType.get())
-        .commit();
+    Assertions.assertThatThrownBy(() -> icebergTable.updateSchema()
+                    .addColumn("data", Types.LongType.get())
+                    .commit())
+            .isInstanceOf(CommitFailedException.class)
+            .hasMessageContaining("is not same as the current table metadata location 'dummylocation'");
   }
 
   @Test
diff --git a/pig/src/test/java/org/apache/iceberg/pig/SchemaUtilTest.java b/pig/src/test/java/org/apache/iceberg/pig/SchemaUtilTest.java
index 41d2336a5d82..bb946ae166c6 100644
--- a/pig/src/test/java/org/apache/iceberg/pig/SchemaUtilTest.java
+++ b/pig/src/test/java/org/apache/iceberg/pig/SchemaUtilTest.java
@@ -34,6 +34,7 @@
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.pig.ResourceSchema;
 import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.assertj.core.api.Assertions;
 import org.junit.Test;
 
 import static org.apache.iceberg.types.Types.NestedField.optional;
@@ -72,11 +73,13 @@ public void testComplex() throws IOException {
     );
   }
 
-  @Test(expected = FrontendException.class)
-  public void invalidMap() throws IOException {
-    convertToPigSchema(new Schema(
-        optional(1, "invalid", MapType.ofOptional(2, 3, IntegerType.get(), DoubleType.get()))
-    ), "", "");
+  @Test
+  public void invalidMap() {
+    Assertions.assertThatThrownBy(() -> convertToPigSchema(new Schema(
+                    optional(1, "invalid", MapType.ofOptional(2, 3, IntegerType.get(), DoubleType.get()))
+            ), "", ""))
+            .isInstanceOf(FrontendException.class)
+            .hasMessageContaining("Unsupported map key type: int");
   }
 
   @Test
diff --git a/spark/v2.4/spark/src/test/java/org/apache/iceberg/examples/SchemaEvolutionTest.java b/spark/v2.4/spark/src/test/java/org/apache/iceberg/examples/SchemaEvolutionTest.java
index afa77f076ec1..a23954ba3f66 100644
--- a/spark/v2.4/spark/src/test/java/org/apache/iceberg/examples/SchemaEvolutionTest.java
+++ b/spark/v2.4/spark/src/test/java/org/apache/iceberg/examples/SchemaEvolutionTest.java
@@ -38,6 +38,7 @@
 import org.apache.spark.sql.types.DataTypes;
 import org.apache.spark.sql.types.LongType$;
 import org.apache.spark.sql.types.StructField;
+import org.assertj.core.api.Assertions;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
@@ -159,14 +160,18 @@ public void updateColumnTypeIntToLong() {
         first.get().dataType() == LongType$.MODULE$);
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void updateColumnTypeIntToString() {
-    table.updateSchema().updateColumn("price", Types.StringType.get()).commit();
+    Assertions.assertThatThrownBy(() -> table.updateSchema().updateColumn("price", Types.StringType.get()).commit())
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Cannot change column type: price: int -> string");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void updateColumnTypeStringToInt() {
-    table.updateSchema().updateColumn("author", Types.IntegerType.get()).commit();
+    Assertions.assertThatThrownBy(() -> table.updateSchema().updateColumn("author", Types.IntegerType.get()).commit())
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Cannot change column type: author: string -> int");
   }
 
   @Test
diff --git a/spark/v2.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java b/spark/v2.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
index 6da76b0b4211..f5e567adb3a5 100644
--- a/spark/v2.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
+++ b/spark/v2.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
@@ -35,6 +35,7 @@
 import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
+import org.assertj.core.api.Assertions;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -176,7 +177,7 @@ public void testSnapshotSelectionByTimestamp() throws IOException {
     Assert.assertEquals("Previous snapshot rows should match", firstBatchRecords, previousSnapshotRecords);
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -184,15 +185,15 @@ public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
     PartitionSpec spec = PartitionSpec.unpartitioned();
     tables.create(SCHEMA, spec, tableLocation);
 
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option("snapshot-id", -10)
-        .load(tableLocation);
-
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option("snapshot-id", -10)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Cannot find snapshot with ID -10");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     long timestamp = System.currentTimeMillis();
 
@@ -201,15 +202,15 @@ public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     PartitionSpec spec = PartitionSpec.unpartitioned();
     tables.create(SCHEMA, spec, tableLocation);
 
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
-
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot find a snapshot older than");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -227,12 +228,13 @@ public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
 
     long timestamp = System.currentTimeMillis();
     long snapshotId = table.currentSnapshot().snapshotId();
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot scan using both snapshot-id and as-of-timestamp");
   }
 }
diff --git a/spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java b/spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
index 6da76b0b4211..22756dd36717 100644
--- a/spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
+++ b/spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
@@ -35,6 +35,7 @@
 import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
+import org.assertj.core.api.Assertions;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -176,7 +177,7 @@ public void testSnapshotSelectionByTimestamp() throws IOException {
     Assert.assertEquals("Previous snapshot rows should match", firstBatchRecords, previousSnapshotRecords);
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -189,10 +190,12 @@ public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
         .option("snapshot-id", -10)
         .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(df::collectAsList)
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Cannot find snapshot with ID -10");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     long timestamp = System.currentTimeMillis();
 
@@ -201,15 +204,15 @@ public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     PartitionSpec spec = PartitionSpec.unpartitioned();
     tables.create(SCHEMA, spec, tableLocation);
 
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
-
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot find a snapshot older than");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -227,12 +230,14 @@ public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
 
     long timestamp = System.currentTimeMillis();
     long snapshotId = table.currentSnapshot().snapshotId();
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot specify both snapshot-id")
+            .hasMessageContaining("and as-of-timestamp");
   }
 }
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
index 6da76b0b4211..22756dd36717 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
@@ -35,6 +35,7 @@
 import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
+import org.assertj.core.api.Assertions;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -176,7 +177,7 @@ public void testSnapshotSelectionByTimestamp() throws IOException {
     Assert.assertEquals("Previous snapshot rows should match", firstBatchRecords, previousSnapshotRecords);
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -189,10 +190,12 @@ public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
         .option("snapshot-id", -10)
         .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(df::collectAsList)
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Cannot find snapshot with ID -10");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     long timestamp = System.currentTimeMillis();
 
@@ -201,15 +204,15 @@ public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     PartitionSpec spec = PartitionSpec.unpartitioned();
     tables.create(SCHEMA, spec, tableLocation);
 
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
-
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot find a snapshot older than");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -227,12 +230,14 @@ public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
 
     long timestamp = System.currentTimeMillis();
     long snapshotId = table.currentSnapshot().snapshotId();
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot specify both snapshot-id")
+            .hasMessageContaining("and as-of-timestamp");
   }
 }
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java
index e1003a175f70..cba5ac686d46 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java
@@ -28,7 +28,7 @@
 
 public class TestHadoopMetricsContextSerialization {
 
-  @Test(expected = Test.None.class)
+  @Test
   public void testHadoopMetricsContextKryoSerialization() throws IOException {
     MetricsContext metricsContext = new HadoopMetricsContext("s3");
 
@@ -41,7 +41,7 @@ public void testHadoopMetricsContextKryoSerialization() throws IOException {
         .increment();
   }
 
-  @Test(expected = Test.None.class)
+  @Test
   public void testHadoopMetricsContextJavaSerialization() throws IOException, ClassNotFoundException {
     MetricsContext metricsContext = new HadoopMetricsContext("s3");
 
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
index 6da76b0b4211..22756dd36717 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
@@ -35,6 +35,7 @@
 import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
+import org.assertj.core.api.Assertions;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -176,7 +177,7 @@ public void testSnapshotSelectionByTimestamp() throws IOException {
     Assert.assertEquals("Previous snapshot rows should match", firstBatchRecords, previousSnapshotRecords);
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -189,10 +190,12 @@ public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
         .option("snapshot-id", -10)
         .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(df::collectAsList)
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Cannot find snapshot with ID -10");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     long timestamp = System.currentTimeMillis();
 
@@ -201,15 +204,15 @@ public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     PartitionSpec spec = PartitionSpec.unpartitioned();
     tables.create(SCHEMA, spec, tableLocation);
 
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
-
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot find a snapshot older than");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -227,12 +230,14 @@ public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
 
     long timestamp = System.currentTimeMillis();
     long snapshotId = table.currentSnapshot().snapshotId();
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot specify both snapshot-id")
+            .hasMessageContaining("and as-of-timestamp");
   }
 }
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java
index e1003a175f70..cba5ac686d46 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/TestHadoopMetricsContextSerialization.java
@@ -28,7 +28,7 @@
 
 public class TestHadoopMetricsContextSerialization {
 
-  @Test(expected = Test.None.class)
+  @Test
   public void testHadoopMetricsContextKryoSerialization() throws IOException {
     MetricsContext metricsContext = new HadoopMetricsContext("s3");
 
@@ -41,7 +41,7 @@ public void testHadoopMetricsContextKryoSerialization() throws IOException {
         .increment();
   }
 
-  @Test(expected = Test.None.class)
+  @Test
   public void testHadoopMetricsContextJavaSerialization() throws IOException, ClassNotFoundException {
     MetricsContext metricsContext = new HadoopMetricsContext("s3");
 
diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
index 6da76b0b4211..22756dd36717 100644
--- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
+++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java
@@ -35,6 +35,7 @@
 import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
+import org.assertj.core.api.Assertions;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -176,7 +177,7 @@ public void testSnapshotSelectionByTimestamp() throws IOException {
     Assert.assertEquals("Previous snapshot rows should match", firstBatchRecords, previousSnapshotRecords);
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -189,10 +190,12 @@ public void testSnapshotSelectionByInvalidSnapshotId() throws IOException {
         .option("snapshot-id", -10)
         .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(df::collectAsList)
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Cannot find snapshot with ID -10");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     long timestamp = System.currentTimeMillis();
 
@@ -201,15 +204,15 @@ public void testSnapshotSelectionByInvalidTimestamp() throws IOException {
     PartitionSpec spec = PartitionSpec.unpartitioned();
     tables.create(SCHEMA, spec, tableLocation);
 
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
-
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot find a snapshot older than");
   }
 
-  @Test(expected = IllegalArgumentException.class)
+  @Test
   public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
     String tableLocation = temp.newFolder("iceberg-table").toString();
 
@@ -227,12 +230,14 @@ public void testSnapshotSelectionBySnapshotIdAndTimestamp() throws IOException {
 
     long timestamp = System.currentTimeMillis();
     long snapshotId = table.currentSnapshot().snapshotId();
-    Dataset<Row> df = spark.read()
-        .format("iceberg")
-        .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
-        .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
-        .load(tableLocation);
 
-    df.collectAsList();
+    Assertions.assertThatThrownBy(() -> spark.read()
+                    .format("iceberg")
+                    .option(SparkReadOptions.SNAPSHOT_ID, snapshotId)
+                    .option(SparkReadOptions.AS_OF_TIMESTAMP, timestamp)
+                    .load(tableLocation))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("Cannot specify both snapshot-id")
+            .hasMessageContaining("and as-of-timestamp");
   }
 }