yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-08-27 10:26:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

MINOR: Fix Create Ingestion when finding an ingestion already (#20243)

Browse Source 
* Fix Create Ingestion when finding an ingestion already

* Add some backoff wait when retrieving pipeline status

* Fix exception
This commit is contained in:
IceS2 2025-03-14 05:06:23 -03:00 committed by GitHub
parent b727f76ce4
commit 312a6dfbdd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 10 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/CreateIngestionPipelineImpl.java
View File

@ -170,7 +170,7 @@ public class CreateIngestionPipelineImpl {
JsonUtils.readOrConvertValue(ingestionPipelineStr, IngestionPipeline.class); JsonUtils.readOrConvertValue(ingestionPipelineStr, IngestionPipeline.class);
if (ingestionPipeline.getPipelineType().equals(pipelineType) if (ingestionPipeline.getPipelineType().equals(pipelineType)
&& ingestionPipeline.getDisplayName().equals(displayName)) { && ingestionPipeline.getDisplayName().equals(displayName)) {
return ingestionPipeline; return ingestionPipeline.withService(service.getEntityReference());
} }
} }
return null; return null;

10
openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/RunIngestionPipelineImpl.java
View File

@ -90,6 +90,7 @@ public class RunIngestionPipelineImpl {
IngestionPipeline ingestionPipeline, IngestionPipeline ingestionPipeline,
long startTime, long startTime,
long timeoutMillis) { long timeoutMillis) {
long backoffMillis = 5 * 1000;
while (true) { while (true) {
if (System.currentTimeMillis() - startTime > timeoutMillis) { if (System.currentTimeMillis() - startTime > timeoutMillis) {
return false; return false;
@ -102,7 +103,14 @@ public class RunIngestionPipelineImpl {
.getData(); .getData();
if (statuses.isEmpty()) { if (statuses.isEmpty()) {
continue; try {
Thread.sleep(backoffMillis);
backoffMillis *= 2;
continue;
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new RuntimeException("Retry interrupted", ie);
}
} }
PipelineStatus status = statuses.get(statuses.size() - 1); PipelineStatus status = statuses.get(statuses.size() - 1);