yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-11-01 11:19:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(restore-indices): add system metadata restoration to restore-indices (#3029)

Browse Source
This commit is contained in:
Gabe Lyons 2021-08-04 11:44:22 -07:00 committed by GitHub
parent 3701931363
commit 2fc71da58d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 25 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/restoreindices/SendMAEStep.java
View File

@ -13,6 +13,7 @@ import com.linkedin.metadata.entity.ebean.EbeanUtils;
import com.linkedin.metadata.models.EntitySpec;
import com.linkedin.metadata.models.registry.EntityRegistry;
import com.linkedin.mxe.MetadataAuditOperation;
import com.linkedin.mxe.SystemMetadata;
import io.ebean.EbeanServer;
import io.ebean.PagedList;
import java.util.Map;
@ -101,13 +102,15 @@ public class SendMAEStep implements UpgradeStep {
return new DefaultUpgradeStepResult(id(), UpgradeStepResult.Result.FAILED);
}
SystemMetadata latestSystemMetadata = EbeanUtils.parseSystemMetadata(aspect.getSystemMetadata());
// 5. Produce MAE events for the aspect record
_entityService.produceMetadataAuditEvent(
urn,
null,
aspectRecord,
null,
null,
latestSystemMetadata,
MetadataAuditOperation.UPDATE
);

10
metadata-io/src/main/java/com/linkedin/metadata/systemmetadata/ElasticSearchSystemMetadataService.java
View File

@ -17,6 +17,7 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.delete.DeleteResponse;
@ -83,9 +84,14 @@ public class ElasticSearchSystemMetadataService implements SystemMetadataService
}
@Override
public void insert(SystemMetadata systemMetadata, String urn, String aspect) {
String document = toDocument(systemMetadata, urn, aspect);
public void insert(@Nullable SystemMetadata systemMetadata, String urn, String aspect) {
if (systemMetadata == null) {
return;
}
String docId = toDocId(urn, aspect);
String document = toDocument(systemMetadata, urn, aspect);
_esDAO.upsertDocument(docId, document);
}

3
metadata-io/src/main/java/com/linkedin/metadata/systemmetadata/SystemMetadataService.java
View File

@ -4,6 +4,7 @@ import com.linkedin.metadata.run.AspectRowSummary;
import com.linkedin.metadata.run.IngestionRunSummary;
import com.linkedin.mxe.SystemMetadata;
import java.util.List;
import javax.annotation.Nullable;
public interface SystemMetadataService {
@ -11,7 +12,7 @@ public interface SystemMetadataService {
void deleteUrn(String finalOldUrn);
void insert(SystemMetadata systemMetadata, String urn, String aspect);
void insert(@Nullable SystemMetadata systemMetadata, String urn, String aspect);
List<AspectRowSummary> findByRunId(String runId);

12
metadata-io/src/test/java/com/linkedin/metadata/systemmetadata/ElasticSearchSystemMetadataServiceTest.java
View File

@ -22,7 +22,6 @@ import org.testng.annotations.Test;
import static org.testng.Assert.*;
public class ElasticSearchSystemMetadataServiceTest {
private ElasticsearchContainer _elasticsearchContainer;
@ -186,4 +185,15 @@ public class ElasticSearchSystemMetadataServiceTest {
assertEquals(rows.size(), 2);
rows.forEach(row -> assertEquals(row.getRunId(), "abc-456"));
}
@Test
public void testInsertNullData() throws Exception {
_client.insert(null, "urn:li:chart:1", "chartKey");
TimeUnit.SECONDS.sleep(5);
List<IngestionRunSummary> runs = _client.listRuns(0, 20);
assertEquals(runs.size(), 0);
}
}