From e03d92e64fe63ee455a4c80b7cfc066bf2ee37ed Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 17 Jul 2026 18:21:22 -0400 Subject: [PATCH 1/2] [improve][test] Add integration tests for the Kinesis source connector Add KinesisSourceIntegrationTest, the first integration test for the Kinesis source. It starts a LocalStack container (Kinesis + DynamoDB + CloudWatch, since the KCL uses DynamoDB for its lease table and CloudWatch for metrics), creates a stream, seeds records, and runs KinesisSource end-to-end, asserting that records flow through the KCL worker into the source queue with the expected value, partition key, and sequence-number property. Mirrors the existing DynamoDBSourceIntegrationTest harness (both are KCL sources against LocalStack): seed before open with TRIM_HORIZON, a background writer for steady supply, a unique applicationName per run for a fresh lease table, and a bounded reader thread because KinesisSource.read() blocks forever on an empty queue. Enhanced fan-out is disabled (useEnhancedFanOut=false) because SubscribeToShard is unreliable on LocalStack. Fixes #122 --- .../kinesis/KinesisSourceIntegrationTest.java | 267 ++++++++++++++++++ 1 file changed, 267 insertions(+) create mode 100644 kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSourceIntegrationTest.java diff --git a/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSourceIntegrationTest.java b/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSourceIntegrationTest.java new file mode 100644 index 0000000000..29d30d0ae1 --- /dev/null +++ b/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSourceIntegrationTest.java @@ -0,0 +1,267 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.kinesis; + +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.io.core.SourceContext; +import org.awaitility.Awaitility; +import org.testcontainers.containers.localstack.LocalStackContainer; +import org.testcontainers.utility.DockerImageName; +import org.testng.annotations.AfterClass; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.core.SdkBytes; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.awssdk.services.kinesis.model.CreateStreamRequest; +import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest; +import software.amazon.awssdk.services.kinesis.model.PutRecordRequest; +import software.amazon.awssdk.services.kinesis.model.StreamStatus; + +/** + * Integration test for {@link KinesisSource} that exercises the full path from a real Kinesis + * stream through the KCL (Kinesis Client Library) worker and into the source's record queue. A + * LocalStack container provides Kinesis, DynamoDB and CloudWatch (the KCL uses DynamoDB for its + * lease table and CloudWatch for metrics). + */ +@Slf4j +public class KinesisSourceIntegrationTest { + + private static final String REGION = "us-east-1"; + private static final String ACCESS_KEY = "access"; + private static final String SECRET_KEY = "secret"; + private static final String STREAM_NAME = "pulsar-kinesis-source-it"; + + // KCL against LocalStack is slow to bootstrap (lease table creation + shard discovery), so the + // first record can take well over a minute to surface. Give each read a generous deadline while + // keeping the overall test bounded so an under-delivering source fails promptly rather than hangs. + private static final int READ_TIMEOUT_SECONDS = 240; + private static final int EXPECTED_RECORDS = 3; + + public static final LocalStackContainer LOCAL_STACK_CONTAINER = + new LocalStackContainer(DockerImageName.parse("localstack/localstack:4.0.3")) + .withServices( + LocalStackContainer.Service.KINESIS, + LocalStackContainer.Service.DYNAMODB, + LocalStackContainer.Service.CLOUDWATCH) + .withEnv("KINESIS_PROVIDER", "kinesalite") + .withStartupTimeout(Duration.ofMinutes(5)); + + private KinesisAsyncClient client; + private KinesisSource source; + private ExecutorService readerExecutor; + private Thread writerThread; + private final AtomicBoolean keepWriting = new AtomicBoolean(false); + + @BeforeClass(alwaysRun = true) + public void beforeClass() { + LOCAL_STACK_CONTAINER.start(); + } + + @AfterClass(alwaysRun = true) + public void afterClass() { + LOCAL_STACK_CONTAINER.stop(); + } + + @BeforeMethod + public void setup() throws Exception { + readerExecutor = Executors.newSingleThreadExecutor(r -> { + Thread t = new Thread(r, "kinesis-it-reader"); + t.setDaemon(true); + return t; + }); + + client = createClient(); + client.createStream(CreateStreamRequest.builder() + .streamName(STREAM_NAME) + .shardCount(1) + .build()).get(); + + Awaitility.await().atMost(60, TimeUnit.SECONDS).until(() -> + client.describeStream(DescribeStreamRequest.builder().streamName(STREAM_NAME).build()) + .get().streamDescription().streamStatus() == StreamStatus.ACTIVE); + log.info("Created Kinesis stream {}", STREAM_NAME); + + source = new KinesisSource(); + } + + @AfterMethod(alwaysRun = true) + public void cleanup() { + keepWriting.set(false); + if (writerThread != null) { + writerThread.interrupt(); + } + if (readerExecutor != null) { + // A reader may still be blocked in read(), which never returns null. + readerExecutor.shutdownNow(); + } + if (source != null) { + try { + source.close(); + } catch (Exception e) { + log.warn("Failed to close source", e); + } + } + if (client != null) { + try { + client.deleteStream(builder -> builder.streamName(STREAM_NAME)).get(); + } catch (Exception e) { + log.warn("Failed to delete stream", e); + } + client.close(); + } + } + + @Test(timeOut = 600_000) + public void testReadFromKinesisStream() throws Exception { + // Seed some records before the source starts; TRIM_HORIZON makes the worker replay them. + for (int i = 0; i < EXPECTED_RECORDS; i++) { + putRecord("seed-" + i); + } + + // Keep writing after open as well so the worker has a steady supply regardless of exactly + // when its lease/shard bootstrap completes. + startBackgroundWriter(); + + SourceContext sourceContext = mock(SourceContext.class); + source.open(buildConfig(), sourceContext); + + int received = 0; + for (int i = 0; i < EXPECTED_RECORDS; i++) { + KinesisRecord record = readOne(); + assertNotNull(record, "read() returned null"); + assertNotNull(record.getValue(), "record had no value"); + assertTrue(record.getValue().length > 0, "record value was empty"); + assertTrue(record.getKey().isPresent(), "record had no key (partition key)"); + assertTrue(record.getProperties().containsKey(KinesisRecord.SEQUENCE_NUMBER), + "record missing SEQUENCE_NUMBER property; got " + record.getProperties()); + log.info("Received Kinesis record: key={} value={} seq={}", + record.getKey().orElse(null), + new String(record.getValue(), StandardCharsets.UTF_8), + record.getProperties().get(KinesisRecord.SEQUENCE_NUMBER)); + received++; + } + assertTrue(received >= EXPECTED_RECORDS, + "Expected at least " + EXPECTED_RECORDS + " records, got " + received); + } + + /** + * Reads a single record on a bounded worker thread, failing (rather than hanging) if none + * arrives in time. {@link KinesisSource#read()} blocks forever on an empty queue, so it must + * never be called on the test thread nor inside an Awaitility assertion. + */ + private KinesisRecord readOne() throws Exception { + Future future = readerExecutor.submit(() -> source.read()); + try { + return future.get(READ_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (TimeoutException e) { + future.cancel(true); + throw new AssertionError("Timed out after " + READ_TIMEOUT_SECONDS + + "s waiting for a record from the Kinesis stream. The source produced no " + + "record; see the KCL/worker logs above.", e); + } + } + + private void startBackgroundWriter() { + keepWriting.set(true); + final AtomicInteger counter = new AtomicInteger(); + writerThread = new Thread(() -> { + while (keepWriting.get()) { + try { + putRecord("live-" + counter.incrementAndGet()); + Thread.sleep(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } catch (Exception e) { + log.warn("Background write failed", e); + } + } + }, "kinesis-it-writer"); + writerThread.setDaemon(true); + writerThread.start(); + } + + private void putRecord(String id) throws Exception { + client.putRecord(PutRecordRequest.builder() + .streamName(STREAM_NAME) + .partitionKey("pk-" + id) + .data(SdkBytes.fromByteArray(("value-" + id).getBytes(StandardCharsets.UTF_8))) + .build()).get(); + } + + private Map buildConfig() { + Map config = new HashMap<>(); + // The source config applies each endpoint via URI.create; LocalStack serves every service on + // one edge port, so the Kinesis endpoint URI works for the DynamoDB (lease table) and + // CloudWatch (metrics) clients too. All three must be set to reach LocalStack. + config.put("awsEndpoint", endpoint()); + config.put("dynamoEndpoint", endpoint()); + config.put("cloudwatchEndpoint", endpoint()); + config.put("awsRegion", REGION); + config.put("awsKinesisStreamName", STREAM_NAME); + config.put("awsCredentialPluginParam", + "{\"accessKey\":\"" + ACCESS_KEY + "\",\"secretKey\":\"" + SECRET_KEY + "\"}"); + // Replay from the start of the stream so the seeded records are delivered. + config.put("initialPositionInStream", "TRIM_HORIZON"); + // Unique application name -> unique KCL lease table per run, avoiding stale leases. + config.put("applicationName", "pulsar-kinesis-it-" + System.currentTimeMillis()); + // Enhanced fan-out (SubscribeToShard) is unreliable on LocalStack; use polling instead. + config.put("useEnhancedFanOut", false); + return config; + } + + private KinesisAsyncClient createClient() { + return KinesisAsyncClient.builder() + .credentialsProvider(new AwsCredentialsProvider() { + @Override + public AwsCredentials resolveCredentials() { + return AwsBasicCredentials.create(ACCESS_KEY, SECRET_KEY); + } + }) + .region(Region.US_EAST_1) + .endpointOverride(LOCAL_STACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.KINESIS)) + .build(); + } + + private static String endpoint() { + final URI uri = LOCAL_STACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.KINESIS); + return uri.toString(); + } +} From 88f6eef2866c7555952107d6917d3be181754ecd Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 17 Jul 2026 19:11:09 -0400 Subject: [PATCH 2/2] [improve][test] Kinesis source IT: address review feedback - Bound the blocking AWS calls in @BeforeMethod (createStream/describeStream) and deleteStream with explicit get() timeouts; @BeforeMethod is not covered by @Test(timeOut=...), so an unbounded future could hang the suite. - Validate that the emitted value and partition key carry what was written (value-* payload, pk-* key) instead of only checking they are non-empty. - Drop the redundant post-loop count assertion (the fixed-count loop already guarantees it). - Use the REGION constant in createClient() instead of hard-coding US_EAST_1. --- .../kinesis/KinesisSourceIntegrationTest.java | 25 +++++++++++-------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSourceIntegrationTest.java b/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSourceIntegrationTest.java index 29d30d0ae1..bac5fb4856 100644 --- a/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSourceIntegrationTest.java +++ b/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSourceIntegrationTest.java @@ -108,14 +108,16 @@ public void setup() throws Exception { }); client = createClient(); + // Bound every blocking AWS call: @BeforeMethod is not covered by @Test(timeOut=...), so an + // unbounded future.get() here could hang the suite if LocalStack stalls. client.createStream(CreateStreamRequest.builder() .streamName(STREAM_NAME) .shardCount(1) - .build()).get(); + .build()).get(30, TimeUnit.SECONDS); Awaitility.await().atMost(60, TimeUnit.SECONDS).until(() -> client.describeStream(DescribeStreamRequest.builder().streamName(STREAM_NAME).build()) - .get().streamDescription().streamStatus() == StreamStatus.ACTIVE); + .get(10, TimeUnit.SECONDS).streamDescription().streamStatus() == StreamStatus.ACTIVE); log.info("Created Kinesis stream {}", STREAM_NAME); source = new KinesisSource(); @@ -140,7 +142,7 @@ public void cleanup() { } if (client != null) { try { - client.deleteStream(builder -> builder.streamName(STREAM_NAME)).get(); + client.deleteStream(builder -> builder.streamName(STREAM_NAME)).get(30, TimeUnit.SECONDS); } catch (Exception e) { log.warn("Failed to delete stream", e); } @@ -162,23 +164,24 @@ public void testReadFromKinesisStream() throws Exception { SourceContext sourceContext = mock(SourceContext.class); source.open(buildConfig(), sourceContext); - int received = 0; for (int i = 0; i < EXPECTED_RECORDS; i++) { KinesisRecord record = readOne(); assertNotNull(record, "read() returned null"); + + // The emitted value and key must carry what we put on the stream: putRecord writes a + // "value-" payload under a "pk-" partition key (id = seed-* or live-*). assertNotNull(record.getValue(), "record had no value"); - assertTrue(record.getValue().length > 0, "record value was empty"); + String value = new String(record.getValue(), StandardCharsets.UTF_8); + assertTrue(value.startsWith("value-"), "unexpected record value: " + value); assertTrue(record.getKey().isPresent(), "record had no key (partition key)"); + assertTrue(record.getKey().get().startsWith("pk-"), + "unexpected partition key: " + record.getKey().get()); assertTrue(record.getProperties().containsKey(KinesisRecord.SEQUENCE_NUMBER), "record missing SEQUENCE_NUMBER property; got " + record.getProperties()); log.info("Received Kinesis record: key={} value={} seq={}", - record.getKey().orElse(null), - new String(record.getValue(), StandardCharsets.UTF_8), + record.getKey().get(), value, record.getProperties().get(KinesisRecord.SEQUENCE_NUMBER)); - received++; } - assertTrue(received >= EXPECTED_RECORDS, - "Expected at least " + EXPECTED_RECORDS + " records, got " + received); } /** @@ -255,7 +258,7 @@ public AwsCredentials resolveCredentials() { return AwsBasicCredentials.create(ACCESS_KEY, SECRET_KEY); } }) - .region(Region.US_EAST_1) + .region(Region.of(REGION)) .endpointOverride(LOCAL_STACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.KINESIS)) .build(); }