[
https://issues.apache.org/jira/browse/BEAM-601?focusedWorklogId=466068&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-466068
]
ASF GitHub Bot logged work on BEAM-601:
---------------------------------------
Author: ASF GitHub Bot
Created on: 04/Aug/20 07:13
Start Date: 04/Aug/20 07:13
Worklog Time Spent: 10m
Work Description: piotr-szuberski commented on a change in pull request
#12422:
URL: https://github.com/apache/beam/pull/12422#discussion_r464846401
##########
File path:
sdks/java/io/kinesis/src/test/java/org/apache/beam/sdk/io/kinesis/KinesisIOIT.java
##########
@@ -99,28 +116,101 @@ private void runRead() {
.withAWSClientsProvider(
options.getAwsAccessKey(),
options.getAwsSecretKey(),
- Regions.fromName(options.getAwsKinesisRegion()))
- .withMaxNumRecords(numberOfRows)
+ Regions.fromName(options.getAwsKinesisRegion()),
+ options.getAwsServiceEndpoint(),
+ options.getAwsVerifyCertificate())
+ .withMaxNumRecords(options.getNumberOfRecords())
// to prevent endless running in case of error
- .withMaxReadTime(Duration.standardMinutes(10))
+ .withMaxReadTime(Duration.standardMinutes(10L))
.withInitialPositionInStream(InitialPositionInStream.AT_TIMESTAMP)
.withInitialTimestampInStream(now)
.withRequestRecordsLimit(1000));
PAssert.thatSingleton(output.apply("Count All", Count.globally()))
- .isEqualTo((long) numberOfRows);
+ .isEqualTo((long) options.getNumberOfRecords());
PCollection<String> consolidatedHashcode =
output
.apply(ParDo.of(new ExtractDataValues()))
.apply("Hash row contents", Combine.globally(new
HashingFn()).withoutDefaults());
PAssert.that(consolidatedHashcode)
- .containsInAnyOrder(TestRow.getExpectedHashForRowCount(numberOfRows));
+
.containsInAnyOrder(TestRow.getExpectedHashForRowCount(options.getNumberOfRecords()));
pipelineRead.run().waitUntilFinish();
}
+ /** Necessary setup for localstack environment. */
+ private static void setupLocalstack() throws Exception {
+
System.setProperty(SDKGlobalConfiguration.DISABLE_CERT_CHECKING_SYSTEM_PROPERTY,
"true");
+
System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY,
"true");
+
+ // For some unclear reason localstack requires the timestamp in seconds
+ now = Instant.ofEpochMilli(Long.divideUnsigned(Instant.now().getMillis(),
1000));
+
+ localstackContainer =
+ new LocalStackContainer("0.11.3")
Review comment:
The default version is 0.8.6, but it is one of those that don't work.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 466068)
Time Spent: 3h 20m (was: 3h 10m)
> Enable Kinesis integration tests
> --------------------------------
>
> Key: BEAM-601
> URL: https://issues.apache.org/jira/browse/BEAM-601
> Project: Beam
> Issue Type: Improvement
> Components: testing
> Affects Versions: 0.3.0-incubating
> Reporter: Przemyslaw Pastuszka
> Assignee: Piotr Szuberski
> Priority: P3
> Labels: starter
> Time Spent: 3h 20m
> Remaining Estimate: 0h
>
> There's an integration test for KinesisIO called KinesisReaderIT, but it is
> currently ignored, because it needs real Kinesis instance setup.
> As part of this task please:
> * setup real Kinesis environment on AWS for testing purposes
> * enable KinesisReaderIT test
> * setup jenkins, so that it passes all KinesisTestOptions when running
> integration tests
> This is a follow up to BEAM-461 requested by [[email protected]] in
> https://github.com/apache/incubator-beam/pull/687/
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
