Skip to content

Commit dadcf2d

Browse files
authored
Re-org reset after backup (#10046)
Make sure that we don't call the update operation of the new scheduler if we are performing a reset operation with the new scheduler being activated.
1 parent a938c4d commit dadcf2d

File tree

5 files changed

+553
-476
lines changed

5 files changed

+553
-476
lines changed

airbyte-server/src/main/java/io/airbyte/server/apis/ConfigurationApi.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -218,7 +218,8 @@ public ConfigurationApi(final ConfigRepository configRepository,
218218
schedulerHandler,
219219
operationsHandler,
220220
featureFlags,
221-
temporalWorkerRunFactory);
221+
temporalWorkerRunFactory,
222+
connectionHelper);
222223
healthCheckHandler = new HealthCheckHandler();
223224
archiveHandler = new ArchiveHandler(
224225
airbyteVersion,

airbyte-server/src/main/java/io/airbyte/server/handlers/ConnectionsHandler.java

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -209,10 +209,17 @@ private Builder<String, Object> generateMetadata(final StandardSync standardSync
209209

210210
public ConnectionRead updateConnection(final ConnectionUpdate connectionUpdate)
211211
throws ConfigNotFoundException, IOException, JsonValidationException {
212+
return updateConnection(connectionUpdate, false);
213+
}
214+
215+
public ConnectionRead updateConnection(final ConnectionUpdate connectionUpdate, boolean isAReset)
216+
throws ConfigNotFoundException, IOException, JsonValidationException {
212217
if (featureFlags.usesNewScheduler()) {
213218
connectionHelper.updateConnection(connectionUpdate);
214219

215-
temporalWorkerRunFactory.update(connectionUpdate);
220+
if (!isAReset) {
221+
temporalWorkerRunFactory.update(connectionUpdate);
222+
}
216223

217224
return connectionHelper.buildConnectionRead(connectionUpdate.getConnectionId());
218225
}

airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java

Lines changed: 19 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
import io.airbyte.commons.lang.MoreBooleans;
4646
import io.airbyte.config.persistence.ConfigNotFoundException;
4747
import io.airbyte.validation.json.JsonValidationException;
48+
import io.airbyte.workers.helper.ConnectionHelper;
4849
import io.airbyte.workers.worker_run.TemporalWorkerRunFactory;
4950
import java.io.IOException;
5051
import java.util.ArrayList;
@@ -71,6 +72,7 @@ public class WebBackendConnectionsHandler {
7172
private final OperationsHandler operationsHandler;
7273
private final FeatureFlags featureFlags;
7374
private final TemporalWorkerRunFactory temporalWorkerRunFactory;
75+
private final ConnectionHelper connectionHelper;
7476

7577
public WebBackendConnectionReadList webBackendListConnectionsForWorkspace(final WorkspaceIdRequestBody workspaceIdRequestBody)
7678
throws ConfigNotFoundException, IOException, JsonValidationException {
@@ -255,23 +257,31 @@ public WebBackendConnectionRead webBackendUpdateConnection(final WebBackendConne
255257
throws ConfigNotFoundException, IOException, JsonValidationException {
256258
final List<UUID> operationIds = updateOperations(webBackendConnectionUpdate);
257259
final ConnectionUpdate connectionUpdate = toConnectionUpdate(webBackendConnectionUpdate, operationIds);
258-
final ConnectionRead connectionRead = connectionsHandler.updateConnection(connectionUpdate);
259260

260-
if (MoreBooleans.isTruthy(webBackendConnectionUpdate.getWithRefreshedCatalog())) {
261-
final ConnectionIdRequestBody connectionId = new ConnectionIdRequestBody().connectionId(webBackendConnectionUpdate.getConnectionId());
262-
263-
if (featureFlags.usesNewScheduler()) {
264-
temporalWorkerRunFactory.synchronousResetConnection(webBackendConnectionUpdate.getConnectionId());
265-
266-
temporalWorkerRunFactory.startNewManualSync(webBackendConnectionUpdate.getConnectionId());
267-
} else {
261+
ConnectionRead connectionRead;
262+
boolean needReset = MoreBooleans.isTruthy(webBackendConnectionUpdate.getWithRefreshedCatalog());
263+
if (!featureFlags.usesNewScheduler()) {
264+
connectionRead = connectionsHandler.updateConnection(connectionUpdate);
265+
if (needReset) {
266+
final ConnectionIdRequestBody connectionId = new ConnectionIdRequestBody().connectionId(webBackendConnectionUpdate.getConnectionId());
268267
// wait for this to execute
269268
schedulerHandler.resetConnection(connectionId);
270269

271270
// just create the job
272271
schedulerHandler.syncConnection(connectionId);
273272
}
273+
} else {
274+
connectionRead = connectionsHandler.updateConnection(connectionUpdate, !needReset);
275+
276+
if (needReset) {
277+
temporalWorkerRunFactory.synchronousResetConnection(webBackendConnectionUpdate.getConnectionId());
278+
279+
temporalWorkerRunFactory.startNewManualSync(webBackendConnectionUpdate.getConnectionId());
280+
281+
connectionRead = connectionHelper.buildConnectionRead(connectionUpdate.getConnectionId());
282+
}
274283
}
284+
275285
return buildWebBackendConnectionRead(connectionRead);
276286
}
277287

0 commit comments

Comments
 (0)