[HUDI-4994] Fix bug that prevents re-ingestion of soft-deleted Datahub entities

hudi-sync/hudi-datahub-sync/src/main/java/org/apache/hudi/sync/datahub/DataHubSyncClient.java

@@ -19,6 +19,7 @@
 
 package org.apache.hudi.sync.datahub;
 
+import com.linkedin.common.Status;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.TableSchemaResolver;
 import org.apache.hudi.common.util.Option;
@@ -52,15 +53,18 @@
 import org.apache.avro.Schema;
 import org.apache.parquet.schema.MessageType;
 
+import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
 public class DataHubSyncClient extends HoodieSyncClient {
 
   protected final DataHubSyncConfig config;
   private final DatasetUrn datasetUrn;
+  private static final Status SOFT_DELETE_FALSE = new Status().setRemoved(false);
 
   public DataHubSyncClient(DataHubSyncConfig config) {
     super(config);
@@ -81,45 +85,39 @@ public void updateLastCommitTimeSynced(String tableName) {
   @Override
   public void updateTableProperties(String tableName, Map<String, String> tableProperties) {
     MetadataChangeProposalWrapper propertiesChangeProposal = MetadataChangeProposalWrapper.builder()
-        .entityType("dataset")
-        .entityUrn(datasetUrn)
-        .upsert()
-        .aspect(new DatasetProperties().setCustomProperties(new StringMap(tableProperties)))
-        .build();
+            .entityType("dataset")
+            .entityUrn(datasetUrn)
+            .upsert()
+            .aspect(new DatasetProperties().setCustomProperties(new StringMap(tableProperties)))
+            .build();
+
+    DatahubResponseLogger responseLogger = new DatahubResponseLogger();
 
     try (RestEmitter emitter = config.getRestEmitter()) {
-      emitter.emit(propertiesChangeProposal, null).get();
+      emitter.emit(propertiesChangeProposal, responseLogger).get();
     } catch (Exception e) {
-      throw new HoodieDataHubSyncException("Fail to change properties for Dataset " + datasetUrn + ": " + tableProperties, e);
+      throw new HoodieDataHubSyncException("Fail to change properties for Dataset " + datasetUrn + ": " +
+              tableProperties, e);
     }
   }
 
   @Override
   public void updateTableSchema(String tableName, MessageType schema) {
     Schema avroSchema = getAvroSchemaWithoutMetadataFields(metaClient);
     List<SchemaField> fields = avroSchema.getFields().stream().map(f -> new SchemaField()
-        .setFieldPath(f.name())
-        .setType(toSchemaFieldDataType(f.schema().getType()))
-        .setDescription(f.doc(), SetMode.IGNORE_NULL)
-        .setNativeDataType(f.schema().getType().getName())).collect(Collectors.toList());
+            .setFieldPath(f.name())
+            .setType(toSchemaFieldDataType(f.schema().getType()))
+            .setDescription(f.doc(), SetMode.IGNORE_NULL)
+            .setNativeDataType(f.schema().getType().getName())).collect(Collectors.toList());
 
     final SchemaMetadata.PlatformSchema platformSchema = new SchemaMetadata.PlatformSchema();
     platformSchema.setOtherSchema(new OtherSchema().setRawSchema(avroSchema.toString()));
-    MetadataChangeProposalWrapper schemaChangeProposal = MetadataChangeProposalWrapper.builder()
-        .entityType("dataset")
-        .entityUrn(datasetUrn)
-        .upsert()
-        .aspect(new SchemaMetadata()
-            .setSchemaName(tableName)
-            .setVersion(0)
-            .setHash("")
-            .setPlatform(datasetUrn.getPlatformEntity())
-            .setPlatformSchema(platformSchema)
-            .setFields(new SchemaFieldArray(fields)))
-        .build();
+    MetadataChangeProposalWrapper schemaChange = createSchemaMetadataUpdate(tableName, fields, platformSchema);
 
+    DatahubResponseLogger responseLogger = new DatahubResponseLogger();
     try (RestEmitter emitter = config.getRestEmitter()) {
-      emitter.emit(schemaChangeProposal, null).get();
+      emitter.emit(schemaChange, responseLogger).get();
+      undoSoftDelete(emitter, responseLogger);
     } catch (Exception e) {
       throw new HoodieDataHubSyncException("Fail to change schema for Dataset " + datasetUrn, e);
     }
@@ -135,6 +133,37 @@ public void close() {
     // no op;
   }
 
+  // When updating an entity, it is ncessary to set its soft-delete status to false, or else the update won't get
+  // reflected in the UI.
+  private void undoSoftDelete(RestEmitter client, DatahubResponseLogger responseLogger) throws IOException, ExecutionException,
+          InterruptedException {
+    MetadataChangeProposalWrapper softDeleteUndoProposal = MetadataChangeProposalWrapper.builder()
+            .entityType("dataset")
+            .entityUrn(datasetUrn)
+            .upsert()
+            .aspect(SOFT_DELETE_FALSE)
+            .aspectName("status")
+            .build();
+
+    client.emit(softDeleteUndoProposal, responseLogger).get();
+  }
+
+  private MetadataChangeProposalWrapper createSchemaMetadataUpdate(String tableName, List<SchemaField> fields,
+                                                                   SchemaMetadata.PlatformSchema platformSchema) {
+    return MetadataChangeProposalWrapper.builder()
+            .entityType("dataset")
+            .entityUrn(datasetUrn)
+            .upsert()
+            .aspect(new SchemaMetadata()
+                    .setSchemaName(tableName)
+                    .setVersion(0)
+                    .setHash("")
+                    .setPlatform(datasetUrn.getPlatformEntity())
+                    .setPlatformSchema(platformSchema)
+                    .setFields(new SchemaFieldArray(fields)))
+            .build();
+  }
+
   static Schema getAvroSchemaWithoutMetadataFields(HoodieTableMetaClient metaClient) {
     try {
       return new TableSchemaResolver(metaClient).getTableAvroSchema(true);

hudi-sync/hudi-datahub-sync/src/main/java/org/apache/hudi/sync/datahub/DatahubResponseLogger.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.sync.datahub;
+
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+/**
+ * Handle responses to requests to Datahub Metastore. Just logs them.
+ */
+public class DatahubResponseLogger implements Callback {
+  private static final Logger LOG = LogManager.getLogger(DatahubResponseLogger.class);
+
+  @Override
+  public void onCompletion(MetadataWriteResponse response) {
+    LOG.info("Completed Datahub RestEmitter request. " +
+            "Status: " + (response.isSuccess() ? " succeeded" : " failed"));
+    if (!response.isSuccess()) {
+      LOG.error("Request failed. " + response);
+    }
+
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Response details: " + response);
+    }
+  }
+
+  @Override
+  public void onFailure(Throwable e) {
+    LOG.error("Error during Datahub RestEmitter request", e);
+  }
+
+}