[HUDI-3304] review

apache · Sep 10, 2022 · cfd03d2 · cfd03d2
1 parent f8963a6
commit cfd03d2
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 5 deletions.
diff --git a/...test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java b/...test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java
@@ -75,8 +75,8 @@ private static Stream<Arguments> writeLogTest() {
     return Stream.of(data).map(Arguments::of);
   }
 
-  private static Stream<Arguments> writeTest() {
-    // enable metadata table, enable embedded time line server
+  private static Stream<Arguments> writePayloadTest() {
+    // Payload class
     Object[] data = new Object[] {
           DefaultHoodieRecordPayload.class.getName(),
           PartialUpdateAvroPayload.class.getName()
@@ -94,7 +94,7 @@ public void setup() {
   }
 
   @ParameterizedTest
-  @MethodSource("writeTest")
+  @MethodSource("writePayloadTest")
   public void testWriteDuringCompaction(String payloadClass) throws IOException {
     HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
         .forTable("test-trip-table")

diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
@@ -424,8 +424,7 @@ private boolean flushBucket(DataBucket bucket) {
     List<HoodieRecord> records = bucket.writeBuffer();
     ValidationUtils.checkState(records.size() > 0, "Data bucket to flush has no buffering records");
     if (config.getBoolean(FlinkOptions.PRE_COMBINE)) {
-      HoodieWriteConfig writeConfig = getHoodieClientConfig(config);
-      records = FlinkWriteHelper.newInstance().deduplicateRecords(records, (HoodieIndex) null, -1, writeConfig.getSchema());
+      records = FlinkWriteHelper.newInstance().deduplicateRecords(records, (HoodieIndex) null, -1, this.writeClient.getConfig().getSchema());
     }
     bucket.preWrite(records);
     final List<WriteStatus> writeStatus = new ArrayList<>(writeFunction.apply(records, instant));