Skip to content

Commit 84cd154

Browse files
alovewalovew
authored
Remove workspace helper from fetchConfigActivity (#21048)
* Remove workspace helper and replace with workspaceApi
1 parent 6c08f43 commit 84cd154

File tree

6 files changed

+67
-65
lines changed

6 files changed

+67
-65
lines changed

‎airbyte-workers/src/main/java/io/airbyte/workers/config/ApiClientBeanFactory.java

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
import io.airbyte.api.client.AirbyteApiClient;
1212
import io.airbyte.api.client.generated.ConnectionApi;
1313
import io.airbyte.api.client.generated.SourceApi;
14+
import io.airbyte.api.client.generated.WorkspaceApi;
1415
import io.airbyte.api.client.invoker.generated.ApiClient;
1516
import io.airbyte.commons.temporal.config.WorkerMode;
1617
import io.micronaut.context.BeanProvider;
@@ -73,6 +74,11 @@ public ConnectionApi connectionApi(final ApiClient apiClient) {
7374
return new ConnectionApi(apiClient);
7475
}
7576

77+
@Singleton
78+
public WorkspaceApi workspaceApi(final ApiClient apiClient) {
79+
return new WorkspaceApi(apiClient);
80+
}
81+
7682
@Singleton
7783
public HttpClient httpClient() {
7884
return HttpClient.newHttpClient();

‎airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionNotificationWorkflowImpl.java

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -9,14 +9,14 @@
99
import io.airbyte.config.Notification;
1010
import io.airbyte.config.Notification.NotificationType;
1111
import io.airbyte.config.SlackNotificationConfiguration;
12-
import io.airbyte.config.StandardSync;
1312
import io.airbyte.config.persistence.ConfigNotFoundException;
1413
import io.airbyte.notification.SlackNotificationClient;
1514
import io.airbyte.validation.json.JsonValidationException;
1615
import io.airbyte.workers.temporal.annotations.TemporalActivityStub;
1716
import io.airbyte.workers.temporal.scheduling.activities.ConfigFetchActivity;
1817
import io.airbyte.workers.temporal.scheduling.activities.NotifySchemaChangeActivity;
1918
import io.airbyte.workers.temporal.scheduling.activities.SlackConfigActivity;
19+
import io.temporal.workflow.Workflow;
2020
import java.io.IOException;
2121
import java.util.Optional;
2222
import java.util.UUID;
@@ -25,6 +25,9 @@
2525
@Slf4j
2626
public class ConnectionNotificationWorkflowImpl implements ConnectionNotificationWorkflow {
2727

28+
private static final String GET_BREAKING_CHANGE_TAG = "get_breaking_change";
29+
private static final int GET_BREAKING_CHANGE_VERSION = 1;
30+
2831
@TemporalActivityStub(activityOptionsBeanName = "shortActivityOptions")
2932
private NotifySchemaChangeActivity notifySchemaChangeActivity;
3033
@TemporalActivityStub(activityOptionsBeanName = "shortActivityOptions")
@@ -35,14 +38,20 @@ public class ConnectionNotificationWorkflowImpl implements ConnectionNotificatio
3538
@Override
3639
public boolean sendSchemaChangeNotification(final UUID connectionId)
3740
throws IOException, InterruptedException, ApiException, ConfigNotFoundException, JsonValidationException {
38-
final StandardSync standardSync = configFetchActivity.getStandardSync(connectionId);
39-
final Optional<SlackNotificationConfiguration> slackConfig = slackConfigActivity.fetchSlackConfiguration(connectionId);
40-
if (slackConfig.isPresent()) {
41-
final Notification notification =
42-
new Notification().withNotificationType(NotificationType.SLACK).withSendOnFailure(false).withSendOnSuccess(false)
43-
.withSlackConfiguration(slackConfig.get());
44-
final SlackNotificationClient notificationClient = new SlackNotificationClient(notification);
45-
return notifySchemaChangeActivity.notifySchemaChange(notificationClient, connectionId, standardSync.getBreakingChange());
41+
final int getBreakingChangeVersion =
42+
Workflow.getVersion(GET_BREAKING_CHANGE_TAG, Workflow.DEFAULT_VERSION, GET_BREAKING_CHANGE_VERSION);
43+
if (getBreakingChangeVersion >= GET_BREAKING_CHANGE_VERSION) {
44+
final Optional<Boolean> breakingChange = configFetchActivity.getBreakingChange(connectionId);
45+
final Optional<SlackNotificationConfiguration> slackConfig = slackConfigActivity.fetchSlackConfiguration(connectionId);
46+
if (slackConfig.isPresent() && breakingChange.isPresent()) {
47+
final Notification notification =
48+
new Notification().withNotificationType(NotificationType.SLACK).withSendOnFailure(false).withSendOnSuccess(false)
49+
.withSlackConfiguration(slackConfig.get());
50+
final SlackNotificationClient notificationClient = new SlackNotificationClient(notification);
51+
return notifySchemaChangeActivity.notifySchemaChange(notificationClient, connectionId, breakingChange.get());
52+
} else {
53+
return false;
54+
}
4655
} else {
4756
return false;
4857
}

‎airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/activities/ConfigFetchActivity.java

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,8 @@
55
package io.airbyte.workers.temporal.scheduling.activities;
66

77
import io.airbyte.api.client.model.generated.ConnectionStatus;
8-
import io.airbyte.config.StandardSync;
9-
import io.airbyte.config.persistence.ConfigNotFoundException;
10-
import io.airbyte.validation.json.JsonValidationException;
118
import io.temporal.activity.ActivityInterface;
129
import io.temporal.activity.ActivityMethod;
13-
import java.io.IOException;
1410
import java.time.Duration;
1511
import java.util.Optional;
1612
import java.util.UUID;
@@ -27,6 +23,9 @@ public interface ConfigFetchActivity {
2723
@ActivityMethod
2824
Optional<ConnectionStatus> getStatus(UUID connectionId);
2925

26+
@ActivityMethod
27+
public Optional<Boolean> getBreakingChange(final UUID connectionId);
28+
3029
@Data
3130
@NoArgsConstructor
3231
@AllArgsConstructor
@@ -45,8 +44,6 @@ class ScheduleRetrieverOutput {
4544

4645
}
4746

48-
StandardSync getStandardSync(final UUID connectionId) throws JsonValidationException, ConfigNotFoundException, IOException;
49-
5047
/**
5148
* Return how much time to wait before running the next sync. It will query the DB to get the last
5249
* starting time of the latest terminal job (Failed, canceled or successful) and return the amount

‎airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/activities/ConfigFetchActivityImpl.java

Lines changed: 24 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import com.google.common.annotations.VisibleForTesting;
1111
import datadog.trace.api.Trace;
1212
import io.airbyte.api.client.generated.ConnectionApi;
13+
import io.airbyte.api.client.generated.WorkspaceApi;
1314
import io.airbyte.api.client.invoker.generated.ApiException;
1415
import io.airbyte.api.client.model.generated.ConnectionIdRequestBody;
1516
import io.airbyte.api.client.model.generated.ConnectionRead;
@@ -19,16 +20,12 @@
1920
import io.airbyte.api.client.model.generated.ConnectionScheduleDataCron;
2021
import io.airbyte.api.client.model.generated.ConnectionScheduleType;
2122
import io.airbyte.api.client.model.generated.ConnectionStatus;
23+
import io.airbyte.api.client.model.generated.WorkspaceRead;
2224
import io.airbyte.commons.temporal.config.WorkerMode;
2325
import io.airbyte.commons.temporal.exception.RetryableException;
24-
import io.airbyte.config.StandardSync;
25-
import io.airbyte.config.persistence.ConfigNotFoundException;
26-
import io.airbyte.config.persistence.ConfigRepository;
2726
import io.airbyte.metrics.lib.ApmTraceUtils;
2827
import io.airbyte.persistence.job.JobPersistence;
29-
import io.airbyte.persistence.job.WorkspaceHelper;
3028
import io.airbyte.persistence.job.models.Job;
31-
import io.airbyte.validation.json.JsonValidationException;
3229
import io.micronaut.context.annotation.Requires;
3330
import io.micronaut.context.annotation.Value;
3431
import jakarta.inject.Named;
@@ -67,43 +64,25 @@ public class ConfigFetchActivityImpl implements ConfigFetchActivity {
6764
UUID.fromString("226edbc1-4a9c-4401-95a9-90435d667d9d"));
6865
private static final long SCHEDULING_NOISE_CONSTANT = 15;
6966

70-
private final ConfigRepository configRepository;
7167
private final JobPersistence jobPersistence;
72-
private final WorkspaceHelper workspaceHelper;
68+
private final WorkspaceApi workspaceApi;
7369
private final Integer syncJobMaxAttempts;
7470
private final Supplier<Long> currentSecondsSupplier;
7571
private final ConnectionApi connectionApi;
7672

77-
public ConfigFetchActivityImpl(final ConfigRepository configRepository,
78-
final JobPersistence jobPersistence,
79-
@Value("${airbyte.worker.sync.max-attempts}") final Integer syncJobMaxAttempts,
80-
@Named("currentSecondsSupplier") final Supplier<Long> currentSecondsSupplier,
81-
final ConnectionApi connectionApi) {
82-
this(configRepository, jobPersistence, new WorkspaceHelper(configRepository, jobPersistence), syncJobMaxAttempts, currentSecondsSupplier,
83-
connectionApi);
84-
}
85-
8673
@VisibleForTesting
87-
protected ConfigFetchActivityImpl(final ConfigRepository configRepository,
88-
final JobPersistence jobPersistence,
89-
final WorkspaceHelper workspaceHelper,
74+
protected ConfigFetchActivityImpl(final JobPersistence jobPersistence,
75+
final WorkspaceApi workspaceApi,
9076
@Value("${airbyte.worker.sync.max-attempts}") final Integer syncJobMaxAttempts,
9177
@Named("currentSecondsSupplier") final Supplier<Long> currentSecondsSupplier,
9278
final ConnectionApi connectionApi) {
93-
this.configRepository = configRepository;
9479
this.jobPersistence = jobPersistence;
95-
this.workspaceHelper = workspaceHelper;
80+
this.workspaceApi = workspaceApi;
9681
this.syncJobMaxAttempts = syncJobMaxAttempts;
9782
this.currentSecondsSupplier = currentSecondsSupplier;
9883
this.connectionApi = connectionApi;
9984
}
10085

101-
@Trace(operationName = ACTIVITY_TRACE_OPERATION_NAME)
102-
@Override
103-
public StandardSync getStandardSync(final UUID connectionId) throws JsonValidationException, ConfigNotFoundException, IOException {
104-
return configRepository.getStandardSync(connectionId);
105-
}
106-
10786
@Trace(operationName = ACTIVITY_TRACE_OPERATION_NAME)
10887
@Override
10988
public ScheduleRetrieverOutput getTimeToWait(final ScheduleRetrieverInput input) {
@@ -176,10 +155,12 @@ private ScheduleRetrieverOutput getTimeToWaitFromScheduleType(final ConnectionRe
176155
}
177156

178157
private Duration addSchedulingNoiseForAllowListedWorkspace(Duration timeToWait, ConnectionRead connectionRead) {
179-
final UUID workspaceId;
158+
UUID workspaceId;
180159
try {
181-
workspaceId = workspaceHelper.getWorkspaceForConnectionId(connectionRead.getConnectionId());
182-
} catch (JsonValidationException | ConfigNotFoundException e) {
160+
ConnectionIdRequestBody connectionIdRequestBody = new ConnectionIdRequestBody().connectionId(connectionRead.getConnectionId());
161+
final WorkspaceRead workspaceRead = workspaceApi.getWorkspaceByConnectionId(connectionIdRequestBody);
162+
workspaceId = workspaceRead.getWorkspaceId();
163+
} catch (ApiException e) {
183164
// We tolerate exceptions and fail open by doing nothing.
184165
return timeToWait;
185166
}
@@ -264,6 +245,19 @@ public Optional<ConnectionStatus> getStatus(final UUID connectionId) {
264245
}
265246
}
266247

248+
@Override
249+
public Optional<Boolean> getBreakingChange(final UUID connectionId) {
250+
try {
251+
final io.airbyte.api.client.model.generated.ConnectionIdRequestBody requestBody =
252+
new io.airbyte.api.client.model.generated.ConnectionIdRequestBody().connectionId(connectionId);
253+
final ConnectionRead connectionRead = connectionApi.getConnection(requestBody);
254+
return Optional.ofNullable(connectionRead.getBreakingChange());
255+
} catch (ApiException e) {
256+
log.info("Encountered an error fetching the connection's breaking change status: ", e);
257+
return Optional.empty();
258+
}
259+
}
260+
267261
private Long getIntervalInSecond(final ConnectionScheduleDataBasicSchedule schedule) {
268262
return getSecondsInUnit(schedule.getTimeUnit()) * schedule.getUnits();
269263
}

‎airbyte-workers/src/test/java/io/airbyte/workers/temporal/scheduling/ConnectionNotificationWorkflowTest.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,6 @@
1313
import io.airbyte.api.client.invoker.generated.ApiException;
1414
import io.airbyte.commons.temporal.scheduling.ConnectionNotificationWorkflow;
1515
import io.airbyte.config.SlackNotificationConfiguration;
16-
import io.airbyte.config.StandardSync;
1716
import io.airbyte.config.persistence.ConfigNotFoundException;
1817
import io.airbyte.notification.SlackNotificationClient;
1918
import io.airbyte.validation.json.JsonValidationException;
@@ -102,7 +101,7 @@ void sendSchemaChangeNotificationNonBreakingChangeTest()
102101

103102
final UUID connectionId = UUID.randomUUID();
104103

105-
when(mConfigFetchActivity.getStandardSync(connectionId)).thenReturn(new StandardSync().withBreakingChange(false));
104+
when(mConfigFetchActivity.getBreakingChange(connectionId)).thenReturn(Optional.of(false));
106105
workflow.sendSchemaChangeNotification(connectionId);
107106

108107
verify(mNotifySchemaChangeActivity, times(1)).notifySchemaChange(any(SlackNotificationClient.class), any(UUID.class), any(boolean.class));

‎airbyte-workers/src/test/java/io/airbyte/workers/temporal/scheduling/activities/ConfigFetchActivityTest.java

Lines changed: 15 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import static org.mockito.Mockito.when;
99

1010
import io.airbyte.api.client.generated.ConnectionApi;
11+
import io.airbyte.api.client.generated.WorkspaceApi;
1112
import io.airbyte.api.client.invoker.generated.ApiException;
1213
import io.airbyte.api.client.model.generated.ConnectionRead;
1314
import io.airbyte.api.client.model.generated.ConnectionSchedule;
@@ -17,10 +18,9 @@
1718
import io.airbyte.api.client.model.generated.ConnectionScheduleDataCron;
1819
import io.airbyte.api.client.model.generated.ConnectionScheduleType;
1920
import io.airbyte.api.client.model.generated.ConnectionStatus;
21+
import io.airbyte.api.client.model.generated.WorkspaceRead;
2022
import io.airbyte.config.persistence.ConfigNotFoundException;
21-
import io.airbyte.config.persistence.ConfigRepository;
2223
import io.airbyte.persistence.job.JobPersistence;
23-
import io.airbyte.persistence.job.WorkspaceHelper;
2424
import io.airbyte.persistence.job.models.Job;
2525
import io.airbyte.validation.json.JsonValidationException;
2626
import io.airbyte.workers.temporal.scheduling.activities.ConfigFetchActivity.ScheduleRetrieverInput;
@@ -46,15 +46,11 @@ class ConfigFetchActivityTest {
4646

4747
private static final Integer SYNC_JOB_MAX_ATTEMPTS = 3;
4848

49-
@Mock
50-
private ConfigRepository mConfigRepository;
51-
5249
@Mock
5350
private JobPersistence mJobPersistence;
5451

5552
@Mock
56-
private WorkspaceHelper mWorkspaceHelper;
57-
53+
private WorkspaceApi mWorkspaceApi;
5854
@Mock
5955
private Job mJob;
6056

@@ -107,7 +103,7 @@ class ConfigFetchActivityTest {
107103
@BeforeEach
108104
void setup() {
109105
configFetchActivity =
110-
new ConfigFetchActivityImpl(mConfigRepository, mJobPersistence, mWorkspaceHelper, SYNC_JOB_MAX_ATTEMPTS,
106+
new ConfigFetchActivityImpl(mJobPersistence, mWorkspaceApi, SYNC_JOB_MAX_ATTEMPTS,
111107
() -> Instant.now().getEpochSecond(), mConnectionApi);
112108
}
113109

@@ -177,7 +173,7 @@ void testDeleted() throws ApiException {
177173
@DisplayName("Test we will wait the required amount of time with legacy config")
178174
void testWait() throws IOException, JsonValidationException, ConfigNotFoundException, ApiException {
179175
configFetchActivity =
180-
new ConfigFetchActivityImpl(mConfigRepository, mJobPersistence, SYNC_JOB_MAX_ATTEMPTS, () -> 60L * 3, mConnectionApi);
176+
new ConfigFetchActivityImpl(mJobPersistence, mWorkspaceApi, SYNC_JOB_MAX_ATTEMPTS, () -> 60L * 3, mConnectionApi);
181177

182178
when(mJob.getStartedAtInSecond())
183179
.thenReturn(Optional.of(60L));
@@ -200,7 +196,7 @@ void testWait() throws IOException, JsonValidationException, ConfigNotFoundExcep
200196
@DisplayName("Test we will not wait if we are late in the legacy schedule schema")
201197
void testNotWaitIfLate() throws IOException, ApiException {
202198
configFetchActivity =
203-
new ConfigFetchActivityImpl(mConfigRepository, mJobPersistence, SYNC_JOB_MAX_ATTEMPTS, () -> 60L * 10, mConnectionApi);
199+
new ConfigFetchActivityImpl(mJobPersistence, mWorkspaceApi, SYNC_JOB_MAX_ATTEMPTS, () -> 60L * 10, mConnectionApi);
204200

205201
when(mJob.getStartedAtInSecond())
206202
.thenReturn(Optional.of(60L));
@@ -255,7 +251,7 @@ void testBasicScheduleTypeFirstRun() throws IOException, ApiException {
255251
@Test
256252
@DisplayName("Test that we will wait the required amount of time with a BASIC_SCHEDULE type on a subsequent run")
257253
void testBasicScheduleSubsequentRun() throws IOException, ApiException {
258-
configFetchActivity = new ConfigFetchActivityImpl(mConfigRepository, mJobPersistence, SYNC_JOB_MAX_ATTEMPTS, () -> 60L * 3, mConnectionApi);
254+
configFetchActivity = new ConfigFetchActivityImpl(mJobPersistence, mWorkspaceApi, SYNC_JOB_MAX_ATTEMPTS, () -> 60L * 3, mConnectionApi);
259255

260256
when(mJob.getStartedAtInSecond())
261257
.thenReturn(Optional.of(60L));
@@ -283,10 +279,10 @@ void testCronScheduleSubsequentRun() throws IOException, JsonValidationException
283279
mockRightNow.set(Calendar.SECOND, 0);
284280
mockRightNow.set(Calendar.MILLISECOND, 0);
285281

286-
when(mWorkspaceHelper.getWorkspaceForConnectionId(any())).thenReturn(UUID.randomUUID());
282+
when(mWorkspaceApi.getWorkspaceByConnectionId(any())).thenReturn(new WorkspaceRead().workspaceId(UUID.randomUUID()));
287283

288284
configFetchActivity =
289-
new ConfigFetchActivityImpl(mConfigRepository, mJobPersistence, mWorkspaceHelper, SYNC_JOB_MAX_ATTEMPTS,
285+
new ConfigFetchActivityImpl(mJobPersistence, mWorkspaceApi, SYNC_JOB_MAX_ATTEMPTS,
290286
() -> mockRightNow.getTimeInMillis() / 1000L, mConnectionApi);
291287

292288
when(mJobPersistence.getLastReplicationJob(connectionId))
@@ -312,10 +308,10 @@ void testCronScheduleMinimumInterval() throws IOException, JsonValidationExcepti
312308
mockRightNow.set(Calendar.SECOND, 0);
313309
mockRightNow.set(Calendar.MILLISECOND, 0);
314310

315-
when(mWorkspaceHelper.getWorkspaceForConnectionId(any())).thenReturn(UUID.randomUUID());
311+
when(mWorkspaceApi.getWorkspaceByConnectionId(any())).thenReturn(new WorkspaceRead().workspaceId(UUID.randomUUID()));
316312

317313
configFetchActivity =
318-
new ConfigFetchActivityImpl(mConfigRepository, mJobPersistence, mWorkspaceHelper, SYNC_JOB_MAX_ATTEMPTS,
314+
new ConfigFetchActivityImpl(mJobPersistence, mWorkspaceApi, SYNC_JOB_MAX_ATTEMPTS,
319315
() -> mockRightNow.getTimeInMillis() / 1000L, mConnectionApi);
320316

321317
when(mJob.getStartedAtInSecond()).thenReturn(Optional.of(mockRightNow.getTimeInMillis() / 1000L));
@@ -342,10 +338,11 @@ void testCronSchedulingNoise() throws IOException, JsonValidationException, Conf
342338
mockRightNow.set(Calendar.SECOND, 0);
343339
mockRightNow.set(Calendar.MILLISECOND, 0);
344340

345-
when(mWorkspaceHelper.getWorkspaceForConnectionId(any())).thenReturn(UUID.fromString("226edbc1-4a9c-4401-95a9-90435d667d9d"));
341+
when(mWorkspaceApi.getWorkspaceByConnectionId(any()))
342+
.thenReturn(new WorkspaceRead().workspaceId(UUID.fromString("226edbc1-4a9c-4401-95a9-90435d667d9d")));
346343

347344
configFetchActivity =
348-
new ConfigFetchActivityImpl(mConfigRepository, mJobPersistence, mWorkspaceHelper, SYNC_JOB_MAX_ATTEMPTS,
345+
new ConfigFetchActivityImpl(mJobPersistence, mWorkspaceApi, SYNC_JOB_MAX_ATTEMPTS,
349346
() -> mockRightNow.getTimeInMillis() / 1000L, mConnectionApi);
350347

351348
when(mJob.getStartedAtInSecond()).thenReturn(Optional.of(mockRightNow.getTimeInMillis() / 1000L));
@@ -371,7 +368,7 @@ class TestGetMaxAttempt {
371368
void testGetMaxAttempt() {
372369
final int maxAttempt = 15031990;
373370
configFetchActivity =
374-
new ConfigFetchActivityImpl(mConfigRepository, mJobPersistence, maxAttempt, () -> Instant.now().getEpochSecond(), mConnectionApi);
371+
new ConfigFetchActivityImpl(mJobPersistence, mWorkspaceApi, maxAttempt, () -> Instant.now().getEpochSecond(), mConnectionApi);
375372
Assertions.assertThat(configFetchActivity.getMaxAttempt().getMaxAttempt())
376373
.isEqualTo(maxAttempt);
377374
}

0 commit comments

Comments
 (0)