This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new 090907a6 [improve][test] Add Aerospike sink integration test (#96)
090907a6 is described below
commit 090907a66f93d8010f7050c5c4f9a6e96982375f
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Fri Jul 10 12:26:08 2026 -0700
[improve][test] Add Aerospike sink integration test (#96)
Adds the first test for the `aerospike` module, an end-to-end
Testcontainers integration test that runs a real Aerospike Community
Edition server, writes records through `AerospikeStringSink`, and reads
them back with an independent `AerospikeClient` to assert the bin values
round-trip. Fixes #47.
Writing exposed a latent defect: `AerospikeAbstractSink` stored values
via `Value.getAsBlob(...)` (Java object serialization), but in the pinned
`aerospike-client-bc:4.5.0`, `BlobValue.estimateSize()` unconditionally
throws "Object serializer has been disabled" -- object serialization was
removed from that client build, so every write failed with result code
-10. Switched the sink to `Value.get(...)`, which stores the value as a
native, first-class Aerospike value (a string bin for the string sink)
that is interoperable and directly queryable.
Container startup is bounded with a two-minute timeout on a log-message
wait strategy so a stalled image pull can never hang CI.
---
aerospike/build.gradle.kts | 6 +
.../pulsar/io/aerospike/AerospikeAbstractSink.java | 2 +-
.../AerospikeStringSinkIntegrationTest.java | 196 +++++++++++++++++++++
3 files changed, 203 insertions(+), 1 deletion(-)
diff --git a/aerospike/build.gradle.kts b/aerospike/build.gradle.kts
index 9fb452f4..9bb17432 100644
--- a/aerospike/build.gradle.kts
+++ b/aerospike/build.gradle.kts
@@ -27,4 +27,10 @@ dependencies {
implementation(libs.jackson.dataformat.yaml)
implementation(libs.aerospike.client)
implementation(libs.bcpkix.jdk18on)
+
+ testImplementation(libs.testcontainers)
+ // The Aerospike 4.5.0 client uses javax.xml.bind.DatatypeConverter
(removed from the JDK
+ // since Java 11) when parsing partition maps. Pulsar's runtime supplies
JAXB in production;
+ // the test classpath must provide it explicitly.
+ testRuntimeOnly(libs.jakarta.xml.bind.api)
}
diff --git
a/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeAbstractSink.java
b/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeAbstractSink.java
index a0cc2809..c3a37227 100644
---
a/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeAbstractSink.java
+++
b/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeAbstractSink.java
@@ -94,7 +94,7 @@ public abstract class AerospikeAbstractSink<K, V> implements
Sink<byte[]> {
KeyValue<K, V> keyValue = extractKeyValue(record);
Key key = new Key(aerospikeSinkConfig.getKeyspace(),
aerospikeSinkConfig.getKeySet(),
keyValue.getKey().toString());
- Bin bin = new Bin(aerospikeSinkConfig.getColumnName(),
Value.getAsBlob(keyValue.getValue()));
+ Bin bin = new Bin(aerospikeSinkConfig.getColumnName(),
Value.get(keyValue.getValue()));
AWriteListener listener = null;
try {
listener = queue.take();
diff --git
a/aerospike/src/test/java/org/apache/pulsar/io/aerospike/AerospikeStringSinkIntegrationTest.java
b/aerospike/src/test/java/org/apache/pulsar/io/aerospike/AerospikeStringSinkIntegrationTest.java
new file mode 100644
index 00000000..ee0772ee
--- /dev/null
+++
b/aerospike/src/test/java/org/apache/pulsar/io/aerospike/AerospikeStringSinkIntegrationTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.aerospike;
+
+import static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import com.aerospike.client.AerospikeClient;
+import com.aerospike.client.Bin;
+import com.aerospike.client.Key;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.SinkContext;
+import org.awaitility.Awaitility;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Integration test for {@link AerospikeStringSink}, exercised against a real
Aerospike
+ * Community Edition server started via Testcontainers.
+ *
+ * <p>Records are pushed through the sink and then read back with an
independent
+ * {@link AerospikeClient} to assert the bin values round-trip correctly. This
is the first
+ * test for the {@code aerospike} module (see
+ * <a href="https://github.com/apache/pulsar-connectors/issues/47";>issue
#47</a>).
+ *
+ * <p>The container startup is bounded with an explicit two-minute timeout on
a log-message wait
+ * strategy so a stalled image pull can never hang the CI job.
+ */
+@Slf4j
+public class AerospikeStringSinkIntegrationTest {
+
+ // Aerospike Community Edition; the default packaged config exposes
namespace "test" on port 3000.
+ private static final DockerImageName AEROSPIKE_IMAGE =
+ DockerImageName.parse("aerospike/aerospike-server:6.4.0.2");
+
+ private static final int AEROSPIKE_PORT = 3000;
+ private static final String NAMESPACE = "test";
+ private static final String SET = "pulsar";
+ private static final String BIN = "value";
+
+ private GenericContainer<?> aerospike;
+ private AerospikeClient client;
+ private AerospikeStringSink sink;
+
+ @BeforeClass(alwaysRun = true)
+ public void setUp() throws Exception {
+ aerospike = new GenericContainer<>(AEROSPIKE_IMAGE)
+ .withExposedPorts(AEROSPIKE_PORT)
+ // "service ready: soon there will be cake!" is logged once
the node is accepting
+ // client requests. Bounded so a stalled pull/start cannot
hang the job.
+ .waitingFor(Wait.forLogMessage(".*service ready.*", 1)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+ aerospike.start();
+
+ final String host = aerospike.getHost();
+ final int port = aerospike.getMappedPort(AEROSPIKE_PORT);
+
+ // Independent client used only to read results back (never to write).
The node logs
+ // "service ready" slightly before its partitions finish rebalancing,
so the first
+ // connection attempts can fail with "node is not yet fully
initialized" -- retry until
+ // the client connects rather than failing fast.
+ Awaitility.await("aerospike client connected")
+ .atMost(60, TimeUnit.SECONDS)
+ .ignoreExceptions()
+ .until(() -> {
+ if (client == null) {
+ client = new AerospikeClient(host, port);
+ }
+ return client.isConnected();
+ });
+
+ // Even once connected, the namespace partition map can still be
settling; a write issued
+ // too early fails with INVALID_NODE ("not yet fully initialized").
Probe with a real
+ // write+read round-trip until the namespace actually accepts
operations, so the sink's
+ // own client (opened next) starts against a fully-initialized cluster.
+ final Key probe = new Key(NAMESPACE, SET, "__probe__");
+ Awaitility.await("aerospike namespace ready")
+ .atMost(60, TimeUnit.SECONDS)
+ .ignoreExceptions()
+ .until(() -> {
+ client.put(null, probe, new Bin(BIN, "ready"));
+ com.aerospike.client.Record r = client.get(null, probe);
+ return r != null && "ready".equals(r.getString(BIN));
+ });
+ client.delete(null, probe);
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("seedHosts", host + ":" + port);
+ config.put("keyspace", NAMESPACE);
+ config.put("keySet", SET);
+ config.put("columnName", BIN);
+
+ sink = new AerospikeStringSink();
+ sink.open(config, mock(SinkContext.class));
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void tearDown() throws Exception {
+ if (sink != null) {
+ try {
+ sink.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ if (client != null) {
+ client.close();
+ }
+ if (aerospike != null) {
+ aerospike.stop();
+ }
+ }
+
+ @Test(timeOut = 300_000)
+ public void testWriteAndReadBack() throws Exception {
+ final int numRecords = 5;
+
+ // Push records through the sink. The sink acks a record only after
the server confirms
+ // the async write, so we use the ack callbacks to know when the data
is durable.
+ CompletableFuture<?>[] futures = new CompletableFuture[numRecords];
+ for (int i = 0; i < numRecords; i++) {
+ final int idx = i;
+ final String key = "key-" + i;
+ final String value = "value-" + i;
+ futures[i] = new CompletableFuture<>();
+ Record<byte[]> record = new Record<byte[]>() {
+ @Override
+ public Optional<String> getKey() {
+ return Optional.of(key);
+ }
+
+ @Override
+ public byte[] getValue() {
+ return value.getBytes(StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public void ack() {
+ futures[idx].complete(null);
+ }
+
+ @Override
+ public void fail() {
+ futures[idx].completeExceptionally(new
RuntimeException("Record " + idx + " failed"));
+ }
+ };
+ sink.write(record);
+ }
+
+ // Wait for every record to be acknowledged (i.e. written) rather than
sleeping.
+ CompletableFuture.allOf(futures).get(60, TimeUnit.SECONDS);
+
+ // Read each record back with an independent client and assert the bin
value round-trips.
+ for (int i = 0; i < numRecords; i++) {
+ final int idx = i;
+ final Key key = new Key(NAMESPACE, SET, "key-" + idx);
+ final String expected = "value-" + idx;
+ Awaitility.await("record key-" + idx + " visible")
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ com.aerospike.client.Record read = client.get(null,
key);
+ assertNotNull(read, "record for key-" + idx + " should
exist in Aerospike");
+ assertEquals(read.getString(BIN), expected,
+ "bin value for key-" + idx + " should match
what was written");
+ });
+ }
+ }
+}
