Skip to content
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

Jump to bottom

don't split lines on LSEP unicode characters when reading lines in destinations #3327

Merged
merged 9 commits into from
May 10, 2021
Merged

don't split lines on LSEP unicode characters when reading lines in destinations #3327

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
6948536
use strict JSONL definition of new lines in destinations
jrhizor May 10, 2021
83541d3
failing test case
jrhizor May 10, 2021
d5bc56d
use next instead of nextLine
jrhizor May 10, 2021
612e4c6
add \n in string for test
jrhizor May 10, 2021
321c426
bump destination versions
jrhizor May 10, 2021
f822040
Merge branch 'master' into jrhizor/skip-lsep-with-scanner
jrhizor May 10, 2021
006a14c
bump to even newer version
jrhizor May 10, 2021
94ae08f
bump versions in dockerfiles as well
jrhizor May 10, 2021
0eaebbc
force mysql test to pass
jrhizor May 10, 2021
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
8 changes: 5 additions & 3 deletions ...rations/bases/base-java/src/main/java/io/airbyte/integrations/base/IntegrationRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,11 +120,13 @@ public void run(String[] args) throws Exception {

@VisibleForTesting
static void consumeWriteStream(AirbyteMessageConsumer consumer) throws Exception {
final Scanner input = new Scanner(System.in);
// use a Scanner that only processes new line characters to strictly abide with the
// https://jsonlines.org/ standard
final Scanner input = new Scanner(System.in).useDelimiter("[\r\n]+");
try (consumer) {
consumer.start();
while (input.hasNextLine()) {
final String inputString = input.nextLine();
while (input.hasNext()) {
final String inputString = input.next();
final Optional<AirbyteMessage> singerMessageOptional = Jsons.tryDeserialize(inputString, AirbyteMessage.class);
if (singerMessageOptional.isPresent()) {
consumer.accept(singerMessageOptional.get());
Expand Down
29 changes: 29 additions & 0 deletions ...-test/src/main/java/io/airbyte/integrations/standardtest/destination/TestDestination.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,6 +348,35 @@ public void testSecondSync() throws Exception {
retrieveRawRecordsAndAssertSameMessages(catalog, secondSyncMessages, defaultSchema);
}

/**
* Tests that we are able to read over special characters properly when processing line breaks in
* destinations.
*/
@Test
public void testLineBreakCharacters() throws Exception {
final AirbyteCatalog catalog =
Jsons.deserialize(MoreResources.readResource(DataArgumentsProvider.EXCHANGE_RATE_CONFIG.catalogFile), AirbyteCatalog.class);
final ConfiguredAirbyteCatalog configuredCatalog = CatalogHelpers.toDefaultConfiguredCatalog(catalog);
final JsonNode config = getConfig();

final List<AirbyteMessage> secondSyncMessages = Lists.newArrayList(new AirbyteMessage()
.withType(Type.RECORD)
.withRecord(new AirbyteRecordMessage()
.withStream(catalog.getStreams().get(0).getName())
.withEmittedAt(Instant.now().toEpochMilli())
.withData(Jsons.jsonNode(ImmutableMap.builder()
.put("id", 1)
.put("currency", "USD\u2028")
.put("date", "2020-03-\n31T00:00:00Z\r")
.put("HKD", 10)
.put("NZD", 700)
.build()))));

runSync(config, secondSyncMessages, configuredCatalog);
final String defaultSchema = getDefaultSchema(config);
retrieveRawRecordsAndAssertSameMessages(catalog, secondSyncMessages, defaultSchema);
}

/**
* Verify that the integration successfully writes records incrementally. The second run should
* append records to the datastore instead of overwriting the previous run.
Expand Down