[gobblin] branch master updated: Fix case where error GTE is incorrectly sent from MCE writer (#3466)

[zi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zihanli58 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git


The following commit(s) were added to refs/heads/master by this push:
     new 5497e54  Fix case where error GTE is incorrectly sent from MCE writer (#3466)
5497e54 is described below

commit 5497e541edc5e1019ebf3437c6e5f244aafdfb60
Author: Jack Moseley <jm...@linkedin.com>
AuthorDate: Fri Feb 11 17:35:01 2022 -0800

    Fix case where error GTE is incorrectly sent from MCE writer (#3466)
---
 .../main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java b/gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
index 9d96eed..c14d7e0 100644
--- a/gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
+++ b/gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
@@ -319,12 +319,12 @@ public class GobblinMCEWriter implements DataWriter<GenericRecord> {
       tableErrorMap.put(tableString, gobblinMetadataException);
     }
     this.datasetErrorMap.put(tableStatus.datasetPath, tableErrorMap);
+    tableOperationTypeMap.remove(tableString);
     log.error(String.format("Meet exception when flush table %s", tableString), e);
     if (datasetErrorMap.size() > maxErrorDataset) {
       //Fail the job if the error size exceeds some number
       throw new IOException(String.format("Container fails to flush for more than %s dataset, last exception we met is: ", maxErrorDataset), e);
     }
-    tableOperationTypeMap.remove(tableString);
   }
 
   // Add fault tolerant ability and make sure we can emit GTE as desired
@@ -358,6 +358,7 @@ public class GobblinMCEWriter implements DataWriter<GenericRecord> {
       this.datasetErrorMap.get(datasetPath).remove(tableString);
     }
   }
+
   /**
    * Call the metadata writers to do flush each table metadata.
    * Flush of metadata writer is the place that do real metadata
@@ -390,7 +391,6 @@ public class GobblinMCEWriter implements DataWriter<GenericRecord> {
 
   @Override
   public void close() throws IOException {
-    this.flush();
     this.closer.close();
   }