piotr-szuberski commented on a change in pull request #12422:
URL: https://github.com/apache/beam/pull/12422#discussion_r466249889
##########
File path:
sdks/java/io/kinesis/src/test/java/org/apache/beam/sdk/io/kinesis/KinesisIOIT.java
##########
@@ -34,34 +42,52 @@
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.joda.time.Duration;
-import org.joda.time.Instant;
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
+import org.testcontainers.containers.localstack.LocalStackContainer;
/**
* Integration test, that writes and reads data to and from real Kinesis. You
need to provide {@link
- * KinesisTestOptions} in order to run this.
+ * KinesisTestOptions} in order to run this if you want to test it with
production setup. By default
+ * when no options are provided an instance of localstack is used.
*/
@RunWith(JUnit4.class)
public class KinesisIOIT implements Serializable {
- private static int numberOfShards;
- private static int numberOfRows;
+ private static final String LOCALSTACK_VERSION = "0.11.3";
@Rule public TestPipeline pipelineWrite = TestPipeline.create();
@Rule public TestPipeline pipelineRead = TestPipeline.create();
+ private static LocalStackContainer localstackContainer;
+ private static String streamName;
+ private static AmazonKinesis kinesisClient;
+
private static KinesisTestOptions options;
- private static final Instant now = Instant.now();
@BeforeClass
- public static void setup() {
+ public static void setup() throws Exception {
PipelineOptionsFactory.register(KinesisTestOptions.class);
options =
TestPipeline.testingPipelineOptions().as(KinesisTestOptions.class);
- numberOfShards = options.getNumberOfShards();
- numberOfRows = options.getNumberOfRecords();
+ if (doUseLocalstack()) {
+ setupLocalstack();
+ }
+ kinesisClient = createKinesisClient();
+ streamName = "beam_test_kinesis" + UUID.randomUUID();
Review comment:
What I meant was to create a new stream just for testing purpose also on
real Kinesis account instead of using an existing one which could have some
data already written (and thus the test's correctness would be violated
assuming that there can be data continously written to an existing stream or
running the test twice at the same moment).
I didn't think about things like different privileges on a pre-created
stream. That's a good point and I agree. I'm not sure who is the target for
this test - is it Beam's CI or a user who wants to test Beam against his
Kinesis account?
----------------------------------------------------------------
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]
