[GitHub] [beam] TheNeuralBit commented on a change in pull request #12422: [BEAM-601] Run KinesisIOIT with localstack

[GitBox]

TheNeuralBit commented on a change in pull request #12422:
URL: https://github.com/apache/beam/pull/12422#discussion_r465389201



##########
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:
       Hm based on that I think what you had before (hard-code "0.11.3") would 
be preferable.That way the test is deterministic.  Let's break it out into a 
constant `LOCALSTACK_VERSION` or something though. 
   
   Sorry for the churn :grimacing: 




----------------------------------------------------------------
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:
us...@infra.apache.org