Skip to content

Bmoric/add non cdc partial reset acceptance test #14764

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 26 commits into from
Jul 18, 2022
Merged

Bmoric/add non cdc partial reset acceptance test #14764

Show file tree
Hide file tree
Changes from 21 commits
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
fd1d874
set per stream feature flag to true for testing
lmossman Jul 8, 2022
c6431de
add a second table to cdc acceptance tests
lmossman Jul 9, 2022
d3d615e
add partial reset test
lmossman Jul 9, 2022
832d944
format
lmossman Jul 9, 2022
36c0e28
add partial reset cdc tests
lmossman Jul 11, 2022
676bfe8
test incremental after partial reset
lmossman Jul 11, 2022
904a7ba
remove dev image from acceptance test
lmossman Jul 12, 2022
42007eb
fix flag and add comment
lmossman Jul 12, 2022
624b747
Revert "set per stream feature flag to true for testing"
lmossman Jul 14, 2022
b04742b
set USE_STREAM_CAPABLE_STATE flag to true in acceptance test script
lmossman Jul 14, 2022
16340d4
call new update endpoint
lmossman Jul 14, 2022
e67ec0e
use methods in test harness instead
lmossman Jul 14, 2022
c71b0ba
remove comment
lmossman Jul 14, 2022
28c6757
add env var to worker container
lmossman Jul 14, 2022
ce315b5
format
lmossman Jul 14, 2022
f5cf0f7
fix state check in basic acceptance test
lmossman Jul 15, 2022
dbd4974
use test info for test name logging
lmossman Jul 15, 2022
5937694
Re-add acceptance test
benmoriceau Jul 15, 2022
16a2196
Re-adda acceptance test
benmoriceau Jul 15, 2022
a375fbf
Format
benmoriceau Jul 15, 2022
9f83989
Merge branch 'master' of github.com:airbytehq/airbyte into bmoric/add…
benmoriceau Jul 15, 2022
70a3415
Merge branch 'master' of github.com:airbytehq/airbyte into bmoric/add…
benmoriceau Jul 18, 2022
9f3639e
Fix acceptance test
benmoriceau Jul 18, 2022
f95ee64
Merge branch 'master' of github.com:airbytehq/airbyte into bmoric/add…
benmoriceau Jul 18, 2022
778a361
Add log
benmoriceau Jul 18, 2022
edb4e20
remove unwanted changes
benmoriceau Jul 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -388,7 +388,6 @@ public WebBackendConnectionRead webBackendUpdateConnectionNew(final WebBackendCo
final ConfiguredAirbyteCatalog existingConfiguredCatalog =
configRepository.getConfiguredCatalogForConnection(connectionId);
ConnectionRead connectionRead;
connectionRead = connectionsHandler.updateConnection(connectionUpdate);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think this change is necessary - since we are capturing the existing configured catalog before we update the connection, it is fine to execute this call here

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reverted, I was wondering why the test were not passing it was not the root cause


final Boolean skipReset = webBackendConnectionUpdate.getSkipReset() != null ? webBackendConnectionUpdate.getSkipReset() : false;
if (!skipReset) {
Expand All @@ -403,6 +402,7 @@ public WebBackendConnectionRead webBackendUpdateConnectionNew(final WebBackendCo
List<io.airbyte.protocol.models.StreamDescriptor> streamsToReset =
allStreamToReset.stream().map(ProtocolConverters::streamDescriptorToProtocol).toList();

connectionRead = connectionsHandler.updateConnection(connectionUpdate);
if (!streamsToReset.isEmpty()) {
final ConnectionIdRequestBody connectionIdRequestBody = new ConnectionIdRequestBody().connectionId(connectionId);
final ConnectionStateType stateType = getStateType(connectionIdRequestBody);
Expand All @@ -418,6 +418,8 @@ public WebBackendConnectionRead webBackendUpdateConnectionNew(final WebBackendCo
verifyManualOperationResult(manualOperationResult);
connectionRead = connectionsHandler.getConnection(connectionUpdate.getConnectionId());
}
} else {
connectionRead = connectionsHandler.updateConnection(connectionUpdate);
}
return buildWebBackendConnectionRead(connectionRead);
}
Expand Down
51 changes: 50 additions & 1 deletion airbyte-test-utils/src/main/java/io/airbyte/test/utils/AirbyteAcceptanceTestHarness.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import io.airbyte.api.client.model.generated.DestinationDefinitionUpdate;
import io.airbyte.api.client.model.generated.DestinationIdRequestBody;
import io.airbyte.api.client.model.generated.DestinationRead;
import io.airbyte.api.client.model.generated.DestinationSyncMode;
import io.airbyte.api.client.model.generated.JobConfigType;
import io.airbyte.api.client.model.generated.JobIdRequestBody;
import io.airbyte.api.client.model.generated.JobListRequestBody;
Expand All @@ -50,6 +51,9 @@
import io.airbyte.api.client.model.generated.SourceDiscoverSchemaRequestBody;
import io.airbyte.api.client.model.generated.SourceIdRequestBody;
import io.airbyte.api.client.model.generated.SourceRead;
import io.airbyte.api.client.model.generated.SyncMode;
import io.airbyte.api.client.model.generated.WebBackendConnectionUpdate;
import io.airbyte.api.client.model.generated.WebBackendOperationCreateOrUpdate;
import io.airbyte.commons.json.Jsons;
import io.airbyte.commons.resources.MoreResources;
import io.airbyte.commons.util.MoreProperties;
Expand Down Expand Up @@ -717,7 +721,7 @@ private void deleteOperation(final UUID destinationId) throws ApiException {
apiClient.getOperationApi().deleteOperation(new OperationIdRequestBody().operationId(destinationId));
}

public JobRead getMostRecentSyncJobId(UUID connectionId) throws Exception {
public JobRead getMostRecentSyncJobId(final UUID connectionId) throws Exception {
return apiClient.getJobsApi()
.listJobsFor(new JobListRequestBody().configId(connectionId.toString()).configTypes(List.of(JobConfigType.SYNC)))
.getJobs()
Expand Down Expand Up @@ -799,4 +803,49 @@ public enum Type {
DESTINATION
}

public void assertDestinationDbEmpty(final boolean withScdTable) throws Exception {
final Database source = getSourceDatabase();
final Set<SchemaTableNamePair> sourceTables = listAllTables(source);
final Set<SchemaTableNamePair> sourceTablesWithRawTablesAdded = addAirbyteGeneratedTables(withScdTable, sourceTables);
final Database destination = getDestinationDatabase();
final Set<SchemaTableNamePair> destinationTables = listAllTables(destination);
assertEquals(sourceTablesWithRawTablesAdded, destinationTables,
String.format("streams did not match.\n source stream names: %s\n destination stream names: %s\n", sourceTables, destinationTables));

for (final SchemaTableNamePair pair : sourceTables) {
final List<JsonNode> sourceRecords = retrieveRawDestinationRecords(pair);
assertTrue(sourceRecords.isEmpty());
}
}

public void setIncrementalAppendSyncMode(final AirbyteCatalog airbyteCatalog, final List<String> cursorField) {
airbyteCatalog.getStreams().forEach(stream -> {
stream.getConfig().syncMode(SyncMode.INCREMENTAL)
.destinationSyncMode(DestinationSyncMode.APPEND)
.cursorField(cursorField);
});
}

public WebBackendConnectionUpdate getUpdateInput(final ConnectionRead connection, final AirbyteCatalog catalog, final OperationRead operation) {
setIncrementalAppendSyncMode(catalog, List.of(COLUMN_ID));

return new WebBackendConnectionUpdate()
.connectionId(connection.getConnectionId())
.name(connection.getName())
.operationIds(connection.getOperationIds())
.operations(List.of(new WebBackendOperationCreateOrUpdate()
.name(operation.getName())
.operationId(operation.getOperationId())
.workspaceId(operation.getWorkspaceId())
.operatorConfiguration(operation.getOperatorConfiguration())))
.namespaceDefinition(connection.getNamespaceDefinition())
.namespaceFormat(connection.getNamespaceFormat())
.syncCatalog(catalog)
.schedule(connection.getSchedule())
.sourceCatalogId(connection.getSourceCatalogId())
.status(connection.getStatus())
.prefix(connection.getPrefix())
.withRefreshedCatalog(true);
}

}
Loading