chore: remove unused activity (#15002)

airbytehq · Jan 14, 2025 · 153436d · 153436d
1 parent 4e02be2
commit 153436d
Show file tree

Hide file tree

Showing 9 changed files with 11 additions and 290 deletions.
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/config/ActivityBeanFactory.java b/airbyte-workers/src/main/java/io/airbyte/workers/config/ActivityBeanFactory.java
@@ -22,7 +22,6 @@
 import io.airbyte.workers.temporal.scheduling.activities.WorkflowConfigActivity;
 import io.airbyte.workers.temporal.sync.AsyncReplicationActivity;
 import io.airbyte.workers.temporal.sync.InvokeOperationsActivity;
-import io.airbyte.workers.temporal.sync.RefreshSchemaActivity;
 import io.airbyte.workers.temporal.sync.ReportRunTimeActivity;
 import io.airbyte.workers.temporal.sync.WorkloadStatusCheckActivity;
 import io.airbyte.workers.temporal.workflows.ConnectorCommandActivity;
@@ -87,14 +86,13 @@ public PayloadChecker payloadChecker(final MetricClient metricClient) {
   @Singleton
   @Named("syncActivities")
   public List<Object> syncActivities(final ConfigFetchActivity configFetchActivity,
-                                     final RefreshSchemaActivity refreshSchemaActivity,
                                      final ReportRunTimeActivity reportRunTimeActivity,
                                      final InvokeOperationsActivity invokeOperationsActivity,
                                      final AsyncReplicationActivity asyncReplicationActivity,
                                      final WorkloadStatusCheckActivity workloadStatusCheckActivity,
                                      final DiscoverCatalogHelperActivity discoverCatalogHelperActivity) {
-    return List.of(configFetchActivity, refreshSchemaActivity, reportRunTimeActivity,
-        invokeOperationsActivity, asyncReplicationActivity, workloadStatusCheckActivity, discoverCatalogHelperActivity);
+    return List.of(configFetchActivity, reportRunTimeActivity, invokeOperationsActivity, asyncReplicationActivity,
+        workloadStatusCheckActivity, discoverCatalogHelperActivity);
   }
 
   @Singleton

diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/RefreshSchemaActivity.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/RefreshSchemaActivity.java
diff --git a/...yte-workers/src/main/java/io/airbyte/workers/temporal/sync/RefreshSchemaActivityImpl.java b/...yte-workers/src/main/java/io/airbyte/workers/temporal/sync/RefreshSchemaActivityImpl.java
diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/SyncWorkflowImpl.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/SyncWorkflowImpl.java
@@ -72,8 +72,6 @@ public class SyncWorkflowImpl implements SyncWorkflow {
 
   private static final HashFunction HASH_FUNCTION = Hashing.md5();
 
-  @TemporalActivityStub(activityOptionsBeanName = "refreshSchemaActivityOptions")
-  private RefreshSchemaActivity refreshSchemaActivity;
   @TemporalActivityStub(activityOptionsBeanName = "shortActivityOptions")
   private ConfigFetchActivity configFetchActivity;
   @TemporalActivityStub(activityOptionsBeanName = "shortActivityOptions")
@@ -119,10 +117,7 @@ public StandardSyncOutput run(final JobRunConfig jobRunConfig,
 
     final Optional<UUID> sourceId = getSourceId(syncInput);
     final RefreshSchemaActivityOutput refreshSchemaOutput;
-    final int removeShouldRereshSchemaVersion = Workflow.getVersion("REMOVE_SHOULD_REFRESH_SCHEMA", DEFAULT_VERSION, 1);
-    final boolean shouldRefreshSchema = sourceId.isPresent() && removeShouldRereshSchemaVersion == DEFAULT_VERSION
-        ? refreshSchemaActivity.shouldRefreshSchema(sourceId.get())
-        : true;
+
     try {
       final JsonNode sourceConfig = configFetchActivity.getSourceConfig(sourceId.get());
       refreshSchemaOutput = runDiscoverAsChildWorkflow(jobRunConfig, sourceLauncherConfig, syncInput, sourceConfig);
@@ -182,12 +177,10 @@ public StandardSyncOutput run(final JobRunConfig jobRunConfig,
               : syncInput.getConnectionContext().getSourceDefinitionId(),
           startTime,
           discoverSchemaEndTime,
-          replicationEndTime,
-          shouldRefreshSchema));
+          replicationEndTime));
     }
 
-    // TODO: remove shouldRefreshSchema on the activity is removed
-    if (shouldRefreshSchema && syncOutput.getStandardSyncSummary() != null && syncOutput.getStandardSyncSummary().getTotalStats() != null) {
+    if (syncOutput.getStandardSyncSummary() != null && syncOutput.getStandardSyncSummary().getTotalStats() != null) {
       syncOutput.getStandardSyncSummary().getTotalStats().setDiscoverSchemaEndTime(discoverSchemaEndTime);
       syncOutput.getStandardSyncSummary().getTotalStats().setDiscoverSchemaStartTime(startTime);
     }

diff --git a/...kers/src/main/kotlin/io/airbyte/workers/temporal/activities/ReportRunTimeActivityInput.kt b/...kers/src/main/kotlin/io/airbyte/workers/temporal/activities/ReportRunTimeActivityInput.kt
@@ -10,7 +10,6 @@ data class ReportRunTimeActivityInput(
   val startTime: Long,
   val refreshSchemaEndTime: Long,
   val replicationEndTime: Long,
-  val shouldRefreshSchema: Boolean,
 ) {
   class Builder
     @JvmOverloads
@@ -20,7 +19,6 @@ data class ReportRunTimeActivityInput(
       private val startTime: Long? = null,
       private val refreshSchemaEndTime: Long? = null,
       private val replicationEndTime: Long? = null,
-      private val shouldRefreshSchema: Boolean? = null,
     ) {
       fun build(): ReportRunTimeActivityInput {
         return ReportRunTimeActivityInput(
@@ -29,7 +27,6 @@ data class ReportRunTimeActivityInput(
           startTime!!,
           refreshSchemaEndTime!!,
           replicationEndTime!!,
-          shouldRefreshSchema!!,
         )
       }
     }

diff --git a/...yte-workers/src/main/kotlin/io/airbyte/workers/temporal/sync/ReportRunTimeActivityImpl.kt b/...yte-workers/src/main/kotlin/io/airbyte/workers/temporal/sync/ReportRunTimeActivityImpl.kt
@@ -20,14 +20,12 @@ class ReportRunTimeActivityImpl(private val metricClient: MetricClient) : Report
     val connectionTag = MetricAttribute(MetricTags.CONNECTION_ID, input.connectionId.toString())
     val sourceDefinitionTag = MetricAttribute(MetricTags.SOURCE_DEFINITION_ID, input.sourceDefinitionId.toString())
 
-    if (input.shouldRefreshSchema) {
-      metricClient.count(
-        OssMetricsRegistry.DISCOVER_CATALOG_RUN_TIME,
-        runTimeRefresh,
-        connectionTag,
-        sourceDefinitionTag,
-      )
-    }
+    metricClient.count(
+      OssMetricsRegistry.DISCOVER_CATALOG_RUN_TIME,
+      runTimeRefresh,
+      connectionTag,
+      sourceDefinitionTag,
+    )
     metricClient.count(OssMetricsRegistry.REPLICATION_RUN_TIME, runTimeReplication, connectionTag, sourceDefinitionTag)
     metricClient.count(OssMetricsRegistry.SYNC_TOTAL_TIME, totalWorkflowRunTime, connectionTag, sourceDefinitionTag)
   }

diff --git a/...src/test/java/io/airbyte/workers/config/DataPlaneActivityInitializationMicronautTest.java b/...src/test/java/io/airbyte/workers/config/DataPlaneActivityInitializationMicronautTest.java
@@ -11,8 +11,6 @@
 import io.airbyte.config.secrets.persistence.SecretPersistence;
 import io.airbyte.workers.temporal.scheduling.activities.ConfigFetchActivity;
 import io.airbyte.workers.temporal.scheduling.activities.ConfigFetchActivityImpl;
-import io.airbyte.workers.temporal.sync.RefreshSchemaActivity;
-import io.airbyte.workers.temporal.sync.RefreshSchemaActivityImpl;
 import io.airbyte.workers.temporal.sync.WebhookOperationActivity;
 import io.airbyte.workers.temporal.sync.WebhookOperationActivityImpl;
 import io.micronaut.context.annotation.Bean;
@@ -57,9 +55,6 @@ class DataPlaneActivityInitializationMicronautTest {
   @Inject
   ConfigFetchActivity configFetchActivity;
 
-  @Inject
-  RefreshSchemaActivity refreshSchemaActivity;
-
   @Inject
   WebhookOperationActivity webhookOperationActivity;
 
@@ -68,11 +63,6 @@ void testConfigFetchActivity() {
     assertEquals(ConfigFetchActivityImpl.class, configFetchActivity.getClass());
   }
 
-  @Test
-  void testRefreshSchemaActivity() {
-    assertEquals(RefreshSchemaActivityImpl.class, refreshSchemaActivity.getClass());
-  }
-
   @Test
   void testWebhookOperationActivity() {
     assertEquals(WebhookOperationActivityImpl.class, webhookOperationActivity.getClass());

diff --git a/...est/java/io/airbyte/workers/temporal/scheduling/activities/RefreshSchemaActivityTest.java b/...est/java/io/airbyte/workers/temporal/scheduling/activities/RefreshSchemaActivityTest.java