jprdonnelly/airbyte
1
0
Fork 0
mirror of synced 2026-01-08 12:03:02 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add Message Migration to Destination Connection Checks (#17954)

Browse Source 
* Add Message Migration to Destination Connection Checks

* Fix test setup
This commit is contained in:
Jimmy Ma
2022-10-17 10:26:41 -07:00
committed by GitHub
parent ef3e84ce3a
commit 5a80c765c2
5 changed files with 32 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
airbyte-server/src/main/java/io/airbyte/server/handlers/SchedulerHandler.java
View File

@@ -177,7 +177,8 @@ public class SchedulerHandler {
final DestinationConnection destination = configRepository.getDestinationConnection(destinationIdRequestBody.getDestinationId());
final StandardDestinationDefinition destinationDef = configRepository.getStandardDestinationDefinition(destination.getDestinationDefinitionId());
final String imageName = DockerUtils.getTaggedImageName(destinationDef.getDockerRepository(), destinationDef.getDockerImageTag());
return reportConnectionStatus(synchronousSchedulerClient.createDestinationCheckConnectionJob(destination, imageName));
final Version protocolVersion = new Version(destinationDef.getProtocolVersion());
return reportConnectionStatus(synchronousSchedulerClient.createDestinationCheckConnectionJob(destination, imageName, protocolVersion));
}
public CheckConnectionRead checkDestinationConnectionFromDestinationCreate(final DestinationCoreConfig destinationConfig)
@@ -195,7 +196,8 @@ public class SchedulerHandler {
.withWorkspaceId(destinationConfig.getWorkspaceId());
final String imageName = DockerUtils.getTaggedImageName(destDef.getDockerRepository(), destDef.getDockerImageTag());
return reportConnectionStatus(synchronousSchedulerClient.createDestinationCheckConnectionJob(destination, imageName));
final Version protocolVersion = new Version(destDef.getProtocolVersion());
return reportConnectionStatus(synchronousSchedulerClient.createDestinationCheckConnectionJob(destination, imageName, protocolVersion));
}
public CheckConnectionRead checkDestinationConnectionFromDestinationIdForUpdate(final DestinationUpdate destinationUpdate)

6
airbyte-server/src/main/java/io/airbyte/server/scheduler/DefaultSynchronousSchedulerClient.java
View File

@@ -87,7 +87,8 @@ public class DefaultSynchronousSchedulerClient implements SynchronousSchedulerCl
@Override
public SynchronousResponse<StandardCheckConnectionOutput> createDestinationCheckConnectionJob(final DestinationConnection destination,
final String dockerImage)
final String dockerImage,
final Version protocolVersion)
throws IOException {
final JsonNode destinationConfiguration = oAuthConfigSupplier.injectDestinationOAuthParameters(
destination.getDestinationDefinitionId(),
@@ -95,7 +96,8 @@ public class DefaultSynchronousSchedulerClient implements SynchronousSchedulerCl
destination.getConfiguration());
final JobCheckConnectionConfig jobCheckConnectionConfig = new JobCheckConnectionConfig()
.withConnectionConfiguration(destinationConfiguration)
.withDockerImage(dockerImage);
.withDockerImage(dockerImage)
.withProtocolVersion(protocolVersion);
final UUID jobId = UUID.randomUUID();
final ConnectorJobReportingContext jobReportingContext = new ConnectorJobReportingContext(jobId, dockerImage);

4
airbyte-server/src/main/java/io/airbyte/server/scheduler/SynchronousSchedulerClient.java
View File

@@ -23,7 +23,9 @@ public interface SynchronousSchedulerClient {
Version protocolVersion)
throws IOException;
SynchronousResponse<StandardCheckConnectionOutput> createDestinationCheckConnectionJob(DestinationConnection destination, String dockerImage)
SynchronousResponse<StandardCheckConnectionOutput> createDestinationCheckConnectionJob(DestinationConnection destination,
String dockerImage,
Version protocolVersion)
throws IOException;
SynchronousResponse<UUID> createDiscoverSchemaJob(SourceConnection source, String dockerImage, String connectorVersion) throws IOException;