-
Notifications
You must be signed in to change notification settings - Fork 3.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge standard sync schedule to standard sync (#3472)
* Copy schedule schema to standard sync * Add config model readme * Remove references of standard sync schedule * Add migration * Fix unit test * Remove standard sync schedule schema * Fix unit test * Remove unnecessary migration resource * Add migration test * Apply spotless format * Add minor updates * Update readme
- Loading branch information
Showing
26 changed files
with
585 additions
and
219 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| # Config Models | ||
|
|
||
| This module uses `jsonschema2pojo` to generate Java config objects from [json schema](https://json-schema.org/) definitions. See [build.gradle](./build.gradle) for details. | ||
|
|
||
| ## How to use | ||
| - Update json schema under: | ||
| ``` | ||
| src/main/resources/types/ | ||
| ``` | ||
| - Run the following command under the project root: | ||
| ```sh | ||
| ./gradlew airbyte-config:models:generateJsonSchema2Pojo | ||
| ``` | ||
| The generated file is under: | ||
| ``` | ||
| build/generated/src/gen/java/io/airbyte/config/ | ||
| ``` | ||
|
|
||
| ## Reference | ||
| - [`jsonschema2pojo` plugin](https://github.com/joelittlejohn/jsonschema2pojo/tree/master/jsonschema2pojo-gradle-plugin). |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
35 changes: 0 additions & 35 deletions
35
airbyte-config/models/src/main/resources/types/StandardSyncSchedule.yaml
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
129 changes: 129 additions & 0 deletions
129
airbyte-migration/src/main/java/io/airbyte/migrate/migrations/MigrationV0_24_0.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,129 @@ | ||
| /* | ||
| * MIT License | ||
| * | ||
| * Copyright (c) 2020 Airbyte | ||
| * | ||
| * Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| * of this software and associated documentation files (the "Software"), to deal | ||
| * in the Software without restriction, including without limitation the rights | ||
| * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| * copies of the Software, and to permit persons to whom the Software is | ||
| * furnished to do so, subject to the following conditions: | ||
| * | ||
| * The above copyright notice and this permission notice shall be included in all | ||
| * copies or substantial portions of the Software. | ||
| * | ||
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
| * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | ||
| * SOFTWARE. | ||
| */ | ||
|
|
||
| package io.airbyte.migrate.migrations; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.fasterxml.jackson.databind.node.ObjectNode; | ||
| import io.airbyte.commons.json.Jsons; | ||
| import io.airbyte.migrate.Migration; | ||
| import io.airbyte.migrate.MigrationUtils; | ||
| import io.airbyte.migrate.ResourceId; | ||
| import io.airbyte.migrate.ResourceType; | ||
| import java.nio.file.Path; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.function.Consumer; | ||
| import java.util.stream.Collectors; | ||
| import java.util.stream.Stream; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| // This migration does the following: | ||
| // 1. Merge StandardSyncSchedule into StandardSync. | ||
| // 2. Remove StandardSyncSchedule. | ||
| public class MigrationV0_24_0 extends BaseMigration implements Migration { | ||
|
|
||
| private static final Logger LOGGER = LoggerFactory.getLogger(MigrationV0_24_0.class); | ||
|
|
||
| protected static final ResourceId STANDARD_SYNC_RESOURCE_ID = ResourceId | ||
| .fromConstantCase(ResourceType.CONFIG, "STANDARD_SYNC"); | ||
| protected static final ResourceId STANDARD_SYNC_SCHEDULE_RESOURCE_ID = ResourceId | ||
| .fromConstantCase(ResourceType.CONFIG, "STANDARD_SYNC_SCHEDULE"); | ||
|
|
||
| private static final String MIGRATION_VERSION = "0.24.0-alpha"; | ||
| private static final Path CONFIG_PATH = Path.of("migrations/migrationV0_24_0"); | ||
|
|
||
| private final Migration previousMigration; | ||
|
|
||
| public MigrationV0_24_0(Migration previousMigration) { | ||
| super(previousMigration); | ||
| this.previousMigration = previousMigration; | ||
| } | ||
|
|
||
| @Override | ||
| public String getVersion() { | ||
| return MIGRATION_VERSION; | ||
| } | ||
|
|
||
| @Override | ||
| public Map<ResourceId, JsonNode> getOutputSchema() { | ||
| final Map<ResourceId, JsonNode> outputSchema = new HashMap<>( | ||
| previousMigration.getOutputSchema()); | ||
| outputSchema.remove(STANDARD_SYNC_SCHEDULE_RESOURCE_ID); | ||
| outputSchema.put( | ||
| STANDARD_SYNC_RESOURCE_ID, | ||
| MigrationUtils.getSchemaFromResourcePath(CONFIG_PATH, STANDARD_SYNC_RESOURCE_ID)); | ||
| return outputSchema; | ||
| } | ||
|
|
||
| @Override | ||
| public void migrate(Map<ResourceId, Stream<JsonNode>> inputData, | ||
| Map<ResourceId, Consumer<JsonNode>> outputData) { | ||
| // Create a map from connection id to standard sync schedule nodes | ||
| // to "join" the schedule onto the standard sync node later. | ||
| final Map<String, JsonNode> connectionToScheduleNodes = inputData | ||
| .get(STANDARD_SYNC_SCHEDULE_RESOURCE_ID) | ||
| .collect(Collectors.toMap(r -> r.get("connectionId").asText(), r -> r)); | ||
|
|
||
| for (final Map.Entry<ResourceId, Stream<JsonNode>> inputEntry : inputData.entrySet()) { | ||
| // Skip standard sync schedule. | ||
| if (inputEntry.getKey().equals(STANDARD_SYNC_SCHEDULE_RESOURCE_ID)) { | ||
| continue; | ||
| } | ||
|
|
||
| inputEntry.getValue().forEach(jsonNode -> { | ||
| if (inputEntry.getKey().equals(STANDARD_SYNC_RESOURCE_ID)) { | ||
| // "Join" the standard sync schedule node onto the standard sync. | ||
| final String connectionId = jsonNode.get("connectionId").asText(); | ||
| final ObjectNode standardSync = (ObjectNode) jsonNode; | ||
| final ObjectNode syncSchedule = (ObjectNode) connectionToScheduleNodes.get(connectionId); | ||
| if (syncSchedule == null) { | ||
| LOGGER.warn( | ||
| "No standard sync schedule config exists for connection {}, will default to manual sync", | ||
| connectionId); | ||
| standardSync.set("manual", Jsons.jsonNode(true)); | ||
| return; | ||
| } | ||
|
|
||
| final JsonNode manual = syncSchedule.get("manual"); | ||
| standardSync.set("manual", manual); | ||
|
|
||
| final JsonNode schedule = syncSchedule.get("schedule"); | ||
| if (schedule != null && !manual.asBoolean()) { | ||
| standardSync.set("schedule", schedule); | ||
| } | ||
|
|
||
| LOGGER.info( | ||
| "Schedule added to standard sync config for connection {} (manual: {}, schedule: {})", | ||
| connectionId, manual, schedule); | ||
| } | ||
|
|
||
| final Consumer<JsonNode> outputConsumer = outputData.get(inputEntry.getKey()); | ||
| outputConsumer.accept(jsonNode); | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| } |
64 changes: 64 additions & 0 deletions
64
airbyte-migration/src/main/resources/migrations/migrationV0_24_0/StandardSync.yaml
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| --- | ||
| "$schema": http://json-schema.org/draft-07/schema# | ||
| "$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/StandardSync.yaml | ||
| title: StandardSync | ||
| description: configuration required for sync for ALL taps | ||
| type: object | ||
| required: | ||
| - sourceId | ||
| - destinationId | ||
| - name | ||
| - catalog | ||
| - manual | ||
| additionalProperties: false | ||
| properties: | ||
| prefix: | ||
| description: Prefix that will be prepended to the name of each stream when it is written to the destination. | ||
| type: string | ||
| sourceId: | ||
| type: string | ||
| format: uuid | ||
| destinationId: | ||
| type: string | ||
| format: uuid | ||
| operationIds: | ||
| type: array | ||
| items: | ||
| type: string | ||
| format: uuid | ||
| connectionId: | ||
| type: string | ||
| format: uuid | ||
| name: | ||
| type: string | ||
| catalog: | ||
| existingJavaType: io.airbyte.protocol.models.ConfiguredAirbyteCatalog | ||
| status: | ||
| type: string | ||
| enum: | ||
| - active | ||
| - inactive | ||
| - deprecated | ||
| # Ideally schedule and manual should be a union, but java | ||
| # codegen does not handle the union type properly. | ||
| # When schedule is defined, manual should be false. | ||
| schedule: | ||
| type: object | ||
| required: | ||
| - timeUnit | ||
| - units | ||
| additionalProperties: false | ||
| properties: | ||
| timeUnit: | ||
| type: string | ||
| enum: | ||
| - minutes | ||
| - hours | ||
| - days | ||
| - weeks | ||
| - months | ||
| units: | ||
| type: integer | ||
| # When manual is true, schedule should be null, and will be ignored. | ||
| manual: | ||
| type: boolean |
Oops, something went wrong.