reuse schemaResolver and remove workaround when last insertion may ge…

…t lost
apache · Jul 1, 2022 · 88ce744 · 88ce744
1 parent 33182d1
commit 88ce744
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 14 deletions.
diff --git a/...datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperator.java b/...datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperator.java
@@ -198,7 +198,8 @@ protected void loadRecords(String partitionPath) throws Exception {
 
     if (latestCommitTime.isPresent()) {
       BaseFileUtils fileUtils = BaseFileUtils.getInstance(this.hoodieTable.getBaseFileFormat());
-      Schema schema = new TableSchemaResolver(this.hoodieTable.getMetaClient()).getTableAvroSchema();
+      TableSchemaResolver schemaResolver = new TableSchemaResolver(this.hoodieTable.getMetaClient());
+      Schema schema = schemaResolver.getTableAvroSchema();
 
       List<FileSlice> fileSlices = this.hoodieTable.getSliceView()
           .getLatestMergedFileSlicesBeforeOrOn(partitionPath, latestCommitTime.get().getTimestamp())
@@ -230,9 +231,8 @@ protected void loadRecords(String partitionPath) throws Exception {
             .filter(logFile -> isValidFile(logFile.getFileStatus()))
             .map(logFile -> logFile.getPath().toString())
             .collect(toList());
-        InternalSchema internalSchema = new TableSchemaResolver(this.hoodieTable.getMetaClient())
-                .getTableInternalSchemaFromCommitMetadata()
-                .orElse(InternalSchema.getEmptyInternalSchema());
+        InternalSchema internalSchema = schemaResolver.getTableInternalSchemaFromCommitMetadata()
+            .orElse(InternalSchema.getEmptyInternalSchema());
         HoodieMergedLogRecordScanner scanner = FormatUtils.logScanner(logPaths, schema, internalSchema, latestCommitTime.get().getTimestamp(),
             writeConfig, hadoopConf);
 

diff --git a/...flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestSchemaEvolution.java b/...flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestSchemaEvolution.java
@@ -296,16 +296,6 @@ private void checkAnswer(TableResult actualResult, String... expectedResult) {
     } catch (Exception e) {
       throw new RuntimeException(e);
     }
-
-    // The last insertion may get lost due to org.apache.hudi.sink.StreamWriteOperatorCoordinator#handleEventFromOperator
-    // can handle EndInputEvent before WriteMetaEvent
-    if (!actual.contains("+I[Julian, 30000.3, 53]")
-        && actual.contains("+I[Julian, null, 53]")
-        && expected.equals(new HashSet<>(Arrays.asList(expectedMergedResult)))) {
-      actual.remove("+I[Julian, null, 53]");
-      actual.add("+I[Julian, 30000.3, 53]");
-    }
-
     assertEquals(expected, actual);
   }