Skip to content

Commit 4f9594d

Browse files
eddumelendezeddumelendez
authored
Add support for apachepulsar/pulsar-all image (#9448)
`apachepulsar/pulsar-all` brings connectors in the image.
1 parent 7c024ed commit 4f9594d

File tree

4 files changed

+113
-61
lines changed

4 files changed

+113
-61
lines changed

‎modules/pulsar/src/main/java/org/testcontainers/containers/PulsarContainer.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
/**
88
* Testcontainers implementation for Apache Pulsar.
99
* <p>
10-
* Supported image: {@code apachepulsar/pulsar}
10+
* Supported images: {@code apachepulsar/pulsar}, {@code apachepulsar/pulsar-all}
1111
* <p>
1212
* Exposed ports:
1313
* <ul>
@@ -64,7 +64,7 @@ public PulsarContainer(String pulsarVersion) {
6464

6565
public PulsarContainer(final DockerImageName dockerImageName) {
6666
super(dockerImageName);
67-
dockerImageName.assertCompatibleWith(DockerImageName.parse("apachepulsar/pulsar"));
67+
dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME, DockerImageName.parse("apachepulsar/pulsar-all"));
6868
withExposedPorts(BROKER_PORT, BROKER_HTTP_PORT);
6969
setWaitStrategy(waitAllStrategy);
7070
}

‎modules/pulsar/src/test/java/org/testcontainers/containers/AbstractPulsar.java‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
package org.testcontainers.containers;
2+
3+
import org.apache.pulsar.client.admin.ListTopicsOptions;
4+
import org.apache.pulsar.client.admin.PulsarAdmin;
5+
import org.apache.pulsar.client.admin.PulsarAdminException;
6+
import org.apache.pulsar.client.api.Consumer;
7+
import org.apache.pulsar.client.api.Message;
8+
import org.apache.pulsar.client.api.Producer;
9+
import org.apache.pulsar.client.api.PulsarClient;
10+
import org.apache.pulsar.client.api.Schema;
11+
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
12+
import org.apache.pulsar.client.api.transaction.Transaction;
13+
14+
import java.util.List;
15+
import java.util.concurrent.CompletableFuture;
16+
import java.util.concurrent.TimeUnit;
17+
18+
import static org.assertj.core.api.Assertions.assertThat;
19+
20+
public class AbstractPulsar {
21+
22+
public static final String TEST_TOPIC = "test_topic";
23+
24+
protected void testPulsarFunctionality(String pulsarBrokerUrl) throws Exception {
25+
try (
26+
PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).build();
27+
Consumer consumer = client.newConsumer().topic(TEST_TOPIC).subscriptionName("test-subs").subscribe();
28+
Producer<byte[]> producer = client.newProducer().topic(TEST_TOPIC).create()
29+
) {
30+
producer.send("test containers".getBytes());
31+
CompletableFuture<Message> future = consumer.receiveAsync();
32+
Message message = future.get(5, TimeUnit.SECONDS);
33+
34+
assertThat(new String(message.getData())).isEqualTo("test containers");
35+
}
36+
}
37+
38+
protected void testTransactionFunctionality(String pulsarBrokerUrl) throws Exception {
39+
try (
40+
PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).enableTransaction(true).build();
41+
Consumer<String> consumer = client
42+
.newConsumer(Schema.STRING)
43+
.topic("transaction-topic")
44+
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
45+
.subscriptionName("test-transaction-sub")
46+
.subscribe();
47+
Producer<String> producer = client
48+
.newProducer(Schema.STRING)
49+
.sendTimeout(0, TimeUnit.SECONDS)
50+
.topic("transaction-topic")
51+
.create()
52+
) {
53+
final Transaction transaction = client.newTransaction().build().get();
54+
producer.newMessage(transaction).value("first").send();
55+
transaction.commit();
56+
Message<String> message = consumer.receive();
57+
assertThat(message.getValue()).isEqualTo("first");
58+
}
59+
}
60+
61+
protected void assertTransactionsTopicCreated(PulsarAdmin pulsarAdmin) throws PulsarAdminException {
62+
final List<String> topics = pulsarAdmin
63+
.topics()
64+
.getPartitionedTopicList("pulsar/system", ListTopicsOptions.builder().includeSystemTopic(true).build());
65+
assertThat(topics).contains("persistent://pulsar/system/transaction_coordinator_assign");
66+
}
67+
}

‎modules/pulsar/src/test/java/org/testcontainers/containers/CompatibleApachePulsarImageTest.java‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
package org.testcontainers.containers;
2+
3+
import org.apache.pulsar.client.admin.PulsarAdmin;
4+
import org.junit.Test;
5+
import org.junit.runner.RunWith;
6+
import org.junit.runners.Parameterized;
7+
import org.testcontainers.utility.DockerImageName;
8+
9+
@RunWith(Parameterized.class)
10+
public class CompatibleApachePulsarImageTest extends AbstractPulsar {
11+
12+
@Parameterized.Parameters(name = "{0}")
13+
public static String[] params() {
14+
return new String[] { "apachepulsar/pulsar:3.0.0", "apachepulsar/pulsar-all:3.0.0" };
15+
}
16+
17+
@Parameterized.Parameter
18+
public String imageName;
19+
20+
@Test
21+
public void testUsage() throws Exception {
22+
try (PulsarContainer pulsar = new PulsarContainer(DockerImageName.parse(this.imageName));) {
23+
pulsar.start();
24+
final String pulsarBrokerUrl = pulsar.getPulsarBrokerUrl();
25+
26+
testPulsarFunctionality(pulsarBrokerUrl);
27+
}
28+
}
29+
30+
@Test
31+
public void testTransactions() throws Exception {
32+
try (PulsarContainer pulsar = new PulsarContainer(DockerImageName.parse(this.imageName)).withTransactions();) {
33+
pulsar.start();
34+
35+
try (PulsarAdmin pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl(pulsar.getHttpServiceUrl()).build()) {
36+
assertTransactionsTopicCreated(pulsarAdmin);
37+
}
38+
testTransactionFunctionality(pulsar.getPulsarBrokerUrl());
39+
}
40+
}
41+
}

‎modules/pulsar/src/test/java/org/testcontainers/containers/PulsarContainerTest.java‎

Lines changed: 3 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -1,29 +1,16 @@
11
package org.testcontainers.containers;
22

3-
import org.apache.pulsar.client.admin.ListTopicsOptions;
43
import org.apache.pulsar.client.admin.PulsarAdmin;
54
import org.apache.pulsar.client.admin.PulsarAdminException;
6-
import org.apache.pulsar.client.api.Consumer;
7-
import org.apache.pulsar.client.api.Message;
8-
import org.apache.pulsar.client.api.Producer;
9-
import org.apache.pulsar.client.api.PulsarClient;
10-
import org.apache.pulsar.client.api.Schema;
11-
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
12-
import org.apache.pulsar.client.api.transaction.Transaction;
135
import org.junit.Test;
146
import org.testcontainers.utility.DockerImageName;
157

168
import java.time.Duration;
17-
import java.util.List;
18-
import java.util.concurrent.CompletableFuture;
19-
import java.util.concurrent.TimeUnit;
209

2110
import static org.assertj.core.api.Assertions.assertThat;
2211
import static org.assertj.core.api.Assertions.assertThatThrownBy;
2312

24-
public class PulsarContainerTest {
25-
26-
public static final String TEST_TOPIC = "test_topic";
13+
public class PulsarContainerTest extends AbstractPulsar {
2714

2815
private static final DockerImageName PULSAR_IMAGE = DockerImageName.parse("apachepulsar/pulsar:3.0.0");
2916

@@ -84,7 +71,8 @@ public void shouldNotEnableFunctionsWorkerByDefault() throws Exception {
8471
public void shouldWaitForFunctionsWorkerStarted() throws Exception {
8572
try (
8673
// constructorWithFunctionsWorker {
87-
PulsarContainer pulsar = new PulsarContainer(PULSAR_IMAGE).withFunctionsWorker();
74+
PulsarContainer pulsar = new PulsarContainer(DockerImageName.parse("apachepulsar/pulsar:3.0.0"))
75+
.withFunctionsWorker();
8876
// }
8977
) {
9078
pulsar.start();
@@ -111,13 +99,6 @@ public void testTransactions() throws Exception {
11199
}
112100
}
113101

114-
private void assertTransactionsTopicCreated(PulsarAdmin pulsarAdmin) throws PulsarAdminException {
115-
final List<String> topics = pulsarAdmin
116-
.topics()
117-
.getPartitionedTopicList("pulsar/system", ListTopicsOptions.builder().includeSystemTopic(true).build());
118-
assertThat(topics).contains("persistent://pulsar/system/transaction_coordinator_assign");
119-
}
120-
121102
@Test
122103
public void testTransactionsAndFunctionsWorker() throws Exception {
123104
try (PulsarContainer pulsar = new PulsarContainer(PULSAR_IMAGE).withTransactions().withFunctionsWorker()) {
@@ -149,41 +130,4 @@ public void testStartupTimeoutIsHonored() {
149130
.hasRootCauseMessage("Precondition failed: timeout must be greater than zero");
150131
}
151132
}
152-
153-
protected void testPulsarFunctionality(String pulsarBrokerUrl) throws Exception {
154-
try (
155-
PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).build();
156-
Consumer consumer = client.newConsumer().topic(TEST_TOPIC).subscriptionName("test-subs").subscribe();
157-
Producer<byte[]> producer = client.newProducer().topic(TEST_TOPIC).create()
158-
) {
159-
producer.send("test containers".getBytes());
160-
CompletableFuture<Message> future = consumer.receiveAsync();
161-
Message message = future.get(5, TimeUnit.SECONDS);
162-
163-
assertThat(new String(message.getData())).isEqualTo("test containers");
164-
}
165-
}
166-
167-
protected void testTransactionFunctionality(String pulsarBrokerUrl) throws Exception {
168-
try (
169-
PulsarClient client = PulsarClient.builder().serviceUrl(pulsarBrokerUrl).enableTransaction(true).build();
170-
Consumer<String> consumer = client
171-
.newConsumer(Schema.STRING)
172-
.topic("transaction-topic")
173-
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
174-
.subscriptionName("test-transaction-sub")
175-
.subscribe();
176-
Producer<String> producer = client
177-
.newProducer(Schema.STRING)
178-
.sendTimeout(0, TimeUnit.SECONDS)
179-
.topic("transaction-topic")
180-
.create()
181-
) {
182-
final Transaction transaction = client.newTransaction().build().get();
183-
producer.newMessage(transaction).value("first").send();
184-
transaction.commit();
185-
Message<String> message = consumer.receive();
186-
assertThat(message.getValue()).isEqualTo("first");
187-
}
188-
}
189133
}

0 commit comments

Comments
 (0)