Fix failing ITs

Author: voonhous

hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java

@@ -25,6 +25,7 @@
 import org.apache.hudi.common.schema.HoodieSchema;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.exception.HoodieAvroSchemaException;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.io.SeekableDataInputStream;
 import org.apache.hudi.storage.HoodieStorage;
@@ -380,10 +381,13 @@ protected HoodieSchema getSchemaFromHeader() {
         (schemaString) -> {
           try {
             return HoodieSchema.parse(schemaStr);
-          } catch (AvroTypeException e) {
+          } catch (HoodieAvroSchemaException e) {
             // Archived commits from earlier hudi versions fail the schema check
-            // So we retry in this one specific instance.
-            return HoodieSchema.parse(schemaStr, false);
+            // So we retry in this one specific instance with validation disabled
+            if (e.getCause() instanceof AvroTypeException) {
+              return HoodieSchema.parse(schemaStr, false);
+            }
+            throw e;
           }
         });
     return SCHEMA_MAP.get(schemaStr);

hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java

@@ -580,7 +580,7 @@ public void testBasicAppendAndRead(HoodieLogBlockType dataBlockType) throws IOEx
         "Read records size should be equal to the written records size");
     assertEquals(convertAvroToSerializableIndexedRecords(copyOfRecords1), recordsRead1,
         "Both records lists should be the same. (ordering guaranteed)");
-    assertEquals(dataBlockRead.getSchema(), getSimpleSchema());
+    assertEquals(dataBlockRead.getSchema().toAvroSchema(), getSimpleSchema());
 
     reader.hasNext();
     nextBlock = reader.next();
@@ -661,7 +661,7 @@ public void testCDCBlock() throws IOException, InterruptedException {
     List<IndexedRecord> recordsRead = getRecords(dataBlockRead);
     assertEquals(3, recordsRead.size(),
         "Read records size should be equal to the written records size");
-    assertEquals(dataBlockRead.getSchema(), cdcSchema);
+    assertEquals(dataBlockRead.getSchema().toAvroSchema(), cdcSchema);
 
     GenericRecord insert = (GenericRecord) recordsRead.stream()
         .filter(record -> record.get(0).toString().equals("i")).findFirst().get();
@@ -2796,7 +2796,7 @@ public void testDataBlockFormatAppendAndReadWithProjectedSchema(
           "Read records size should be equal to the written records size");
       assertEquals(expectedRecords, recordsRead,
           "Both records lists should be the same. (ordering guaranteed)");
-      assertEquals(dataBlockRead.getSchema(), projectedSchema);
+      assertEquals(dataBlockRead.getSchema().toAvroSchema(), projectedSchema);
 
       int bytesRead = (int) BenchmarkCounter.getBytesRead();