Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Tests: use pulsar server in docker #120

Merged
merged 11 commits into from
Apr 9, 2024
Merged

Tests: use pulsar server in docker #120

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
bcb6fc6
make build compatible with jdk21
nicoloboschi Apr 8, 2024
a703872
tests: use pulsar server in docker
nicoloboschi Apr 9, 2024
e095378
spotbugs
nicoloboschi Apr 9, 2024
f8b8e60
fix
nicoloboschi Apr 9, 2024
00eb4e2
fix
nicoloboschi Apr 9, 2024
2475802
fix test
nicoloboschi Apr 9, 2024
d7ba980
fix test
nicoloboschi Apr 9, 2024
550ad16
add ssh
nicoloboschi Apr 9, 2024
d790331
timeout
nicoloboschi Apr 9, 2024
ca695f1
Fix test and also wait for Pulsar to be ready
eolivelli Apr 9, 2024
3345d0c
fix log
nicoloboschi Apr 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion examples/spring/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.20</version>
<version>1.18.30</version>
<scope>provided</scope>
</dependency>
</dependencies>
Expand Down
26 changes: 15 additions & 11 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,15 +61,15 @@
<hawtbuf.version>1.11</hawtbuf.version>
<curator.version>5.1.0</curator.version>
<slf4j.version>1.7.30</slf4j.version>
<logback.version>1.2.3</logback.version>
<junit.version>5.7.1</junit.version>
<surefire.version>3.0.0-M5</surefire.version>
<surefire.version>3.1.0</surefire.version>
<jackson.version>2.14.2</jackson.version>
<gson.version>2.8.9</gson.version>
<commons-compress.version>1.21</commons-compress.version>
<awaitility.version>4.0.3</awaitility.version>
<maven.antrun.plugin.version>3.1.0</maven.antrun.plugin.version>
<snakeyaml.version>2.0</snakeyaml.version>
<testcontainers.version>1.18.3</testcontainers.version>
<!-- required for running tests on JDK11+ -->
<test.additional.args>--add-opens java.base/java.lang.reflect=ALL-UNNAMED --add-opens java.base/jdk.internal.loader=ALL-UNNAMED --add-opens java.base/java.lang=ALL-UNNAMED
<!--Mockito-->--add-opens java.base/java.io=ALL-UNNAMED
Expand Down Expand Up @@ -125,7 +125,7 @@
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.20</version>
<version>1.18.30</version>
</dependency>
<dependency>
<groupId>${pulsar.groupId}</groupId>
Expand Down Expand Up @@ -214,6 +214,13 @@
<artifactId>snakeyaml</artifactId>
<version>${snakeyaml.version}</version>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-bom</artifactId>
<version>${testcontainers.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
Expand All @@ -232,7 +239,7 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.1</version>
<version>3.12.1</version>
<configuration>
<compilerArgs>-Xlint:unchecked</compilerArgs>
<fork>false</fork>
Expand Down Expand Up @@ -320,25 +327,22 @@ limitations under the License.]]></inlineHeader>
<configuration>
<argLine>${test.additional.args}</argLine>
<trimStackTrace>false</trimStackTrace>
<systemPropertyVariables>
<logback.configurationFile>${project.basedir}/src/test/resources/logback-test.xml</logback.configurationFile>
</systemPropertyVariables>
</configuration>
</plugin>
<plugin>
<artifactId>maven-failsafe-plugin</artifactId>
<version>${surefire.version}</version>
<configuration>
<trimStackTrace>false</trimStackTrace>
<systemPropertyVariables>
<logback.configurationFile>${project.basedir}/src/test/resources/logback-test.xml</logback.configurationFile>
</systemPropertyVariables>
</configuration>
</plugin>
<plugin>
<groupId>com.github.spotbugs</groupId>
<artifactId>spotbugs-maven-plugin</artifactId>
<version>4.2.2</version>
<version>4.8.3.1</version>
<configuration>
<omitVisitors>FindReturnRef,ConstructorThrow</omitVisitors>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
Expand Down
23 changes: 14 additions & 9 deletions pulsar-jms/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,23 +88,28 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>${pulsar.groupId}</groupId>
<artifactId>pulsar-broker</artifactId>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
eolivelli marked this conversation as resolved.
Show resolved Hide resolved
<artifactId>pulsar</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
Expand All @@ -124,7 +129,7 @@
<target>
<echo>copy filters</echo>
<mkdir dir="${project.build.outputDirectory}/filters"/>
<copy verbose="true" file="${basedir}/../pulsar-jms-filters/target/pulsar-jms-filters-${project.version}.jar" tofile="${project.build.outputDirectory}/filters/jms-filter.nar"/>
<copy verbose="true" file="${basedir}/../pulsar-jms-filters/target/pulsar-jms-${project.version}-nar.nar" tofile="${project.build.outputDirectory}/filters/jms-filter.nar"/>
</target>
</configuration>
</execution>
Expand Down
1 change: 1 addition & 0 deletions pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/messages/PulsarStreamMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -403,6 +403,7 @@ public short readShort() throws JMSException {
* @throws MessageFormatException if this type conversion is invalid
* @throws MessageNotReadableException if the message is in write-only mode.
*/
@SuppressFBWarnings("DCN_NULLPOINTER_EXCEPTION")
public char readChar() throws JMSException {
checkReadable();
try {
Expand Down
45 changes: 11 additions & 34 deletions pulsar-jms/src/test/java/com/datastax/oss/pulsar/jms/AcknowledgementModeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,7 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

import com.datastax.oss.pulsar.jms.utils.PulsarCluster;
import java.nio.file.Path;
import com.datastax.oss.pulsar.jms.utils.PulsarContainerExtension;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
Expand All @@ -38,36 +37,20 @@
import javax.jms.TextMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.api.extension.RegisterExtension;

@Timeout(30)
@Slf4j
public class AcknowledgementModeTest {

@TempDir public static Path tempDir;
private static PulsarCluster cluster;

@BeforeAll
public static void before() throws Exception {
cluster = new PulsarCluster(tempDir);
cluster.start();
}

@AfterAll
public static void after() throws Exception {
if (cluster != null) {
cluster.close();
}
}
@RegisterExtension
static PulsarContainerExtension pulsarContainer = new PulsarContainerExtension();

@Test
public void testAUTO_ACKNOWLEDGE() throws Exception {
Map<String, Object> properties = new HashMap<>();
properties.put("webServiceUrl", cluster.getAddress());
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
try (PulsarConnectionFactory factory = new PulsarConnectionFactory(properties); ) {
try (Connection connection = factory.createConnection()) {
connection.start();
Expand Down Expand Up @@ -97,8 +80,7 @@ public void testAUTO_ACKNOWLEDGE() throws Exception {

@Test
public void testAUTO_ACKNOWLEDGE_ackReceipt() throws Exception {
Map<String, Object> properties = new HashMap<>();
properties.put("webServiceUrl", cluster.getAddress());
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
Map<String, Object> consumerConfig = new HashMap<>();
consumerConfig.put("ackReceiptEnabled", true);
properties.put("consumerConfig", consumerConfig);
Expand Down Expand Up @@ -145,8 +127,7 @@ public void onException(Message message, Exception e) {}

@Test
public void testADUPS_OK_ACKNOWLEDGE() throws Exception {
Map<String, Object> properties = new HashMap<>();
properties.put("webServiceUrl", cluster.getAddress());
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
try (PulsarConnectionFactory factory = new PulsarConnectionFactory(properties); ) {
try (Connection connection = factory.createConnection()) {
connection.start();
Expand Down Expand Up @@ -176,8 +157,7 @@ public void testADUPS_OK_ACKNOWLEDGE() throws Exception {

@Test()
public void testACLIENT_ACKNOWLEDGE() throws Exception {
Map<String, Object> properties = new HashMap<>();
properties.put("webServiceUrl", cluster.getAddress());
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
try (PulsarConnectionFactory factory = new PulsarConnectionFactory(properties); ) {
try (Connection connection = factory.createConnection()) {
connection.start();
Expand Down Expand Up @@ -225,8 +205,7 @@ public void testACLIENT_ACKNOWLEDGE() throws Exception {

@Test
public void testINDIVIDUAL_ACKNOWLEDGE() throws Exception {
Map<String, Object> properties = new HashMap<>();
properties.put("webServiceUrl", cluster.getAddress());
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
try (PulsarConnectionFactory factory = new PulsarConnectionFactory(properties); ) {
try (Connection connection = factory.createConnection()) {
connection.start();
Expand Down Expand Up @@ -290,8 +269,7 @@ public void testINDIVIDUAL_ACKNOWLEDGEWithBatchingWithoutBatchIndexAckEnabled()

private void testINDIVIDUAL_ACKNOWLEDGEWithBatching(boolean batchIndexAckEnabled)
throws Exception {
Map<String, Object> properties = new HashMap<>();
properties.put("webServiceUrl", cluster.getAddress());
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
Map<String, Object> producerConfig = new HashMap<>();
producerConfig.put("batchingEnabled", "true");
producerConfig.put("batchingMaxPublishDelayMicros", "1000000");
Expand Down Expand Up @@ -403,8 +381,7 @@ public void onException(Message message, Exception e) {}

@Test
public void testAutoNackWrongType() throws Exception {
Map<String, Object> properties = new HashMap<>();
properties.put("webServiceUrl", cluster.getAddress());
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
try (PulsarConnectionFactory factory = new PulsarConnectionFactory(properties); ) {
try (JMSContext session = factory.createContext()) {
Queue destination =
Expand Down
76 changes: 23 additions & 53 deletions pulsar-jms/src/test/java/com/datastax/oss/pulsar/jms/BasicServerSideFilterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,7 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.*;

import com.datastax.oss.pulsar.jms.utils.PulsarCluster;
import java.nio.file.Path;
import com.datastax.oss.pulsar.jms.utils.PulsarContainerExtension;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
Expand All @@ -40,45 +39,24 @@
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.internal.util.reflection.Whitebox;

@Slf4j
public class BasicServerSideFilterTest {

@TempDir public static Path tempDir;
private static PulsarCluster cluster;

public BasicServerSideFilterTest() {}

@BeforeAll
public static void before() throws Exception {
cluster =
new PulsarCluster(
tempDir,
(config) -> {
config.setTransactionCoordinatorEnabled(false);
config.setAllowAutoTopicCreation(false);
});
cluster.start();
}

@AfterAll
public static void after() throws Exception {
if (cluster != null) {
cluster.close();
}
}
@RegisterExtension
static PulsarContainerExtension pulsarContainer =
new PulsarContainerExtension()
.withEnv("PULSAR_PREFIX_transactionCoordinatorEnabled", "false")
.withEnv("PULSAR_PREFIX_allowAutoTopicCreation", "false");

static int refreshServerSideFiltersPeriod = 10;

private Map<String, Object> buildProperties() {
Map<String, Object> properties = new HashMap<>();
properties.put("webServiceUrl", cluster.getAddress());
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();

properties.put("jms.useServerSideFiltering", "true");
properties.put("jms.refreshServerSideFiltersPeriod", refreshServerSideFiltersPeriod);
Expand All @@ -105,15 +83,11 @@ public void downloadSubscriptionProperties(int numPartitions) throws Exception {
String topicName = "topic-with-sub-" + UUID.randomUUID();
String topicName2 = "topic-with-sub-" + UUID.randomUUID();
if (numPartitions > 0) {
cluster
.getService()
.getAdminClient()
.topics()
.createPartitionedTopic(topicName, numPartitions);

cluster
.getService()
.getAdminClient()

pulsarContainer.getAdmin().topics().createPartitionedTopic(topicName, numPartitions);

pulsarContainer
.getAdmin()
.namespaces()
.setAutoTopicCreation(
"public/default",
Expand All @@ -123,10 +97,9 @@ public void downloadSubscriptionProperties(int numPartitions) throws Exception {
.allowAutoTopicCreation(true)
.build());
} else {
cluster.getService().getAdminClient().topics().createNonPartitionedTopic(topicName);
cluster
.getService()
.getAdminClient()
pulsarContainer.getAdmin().topics().createNonPartitionedTopic(topicName);
pulsarContainer
.getAdmin()
.namespaces()
.setAutoTopicCreation(
"public/default",
Expand All @@ -144,9 +117,8 @@ public void downloadSubscriptionProperties(int numPartitions) throws Exception {
subscriptionProperties.put("jms.filtering", "true");

// create a Subscription with a selector
cluster
.getService()
.getAdminClient()
pulsarContainer
.getAdmin()
.topics()
.createSubscription(
topicName, subscriptionName, MessageId.earliest, false, subscriptionProperties);
Expand All @@ -172,7 +144,7 @@ public void downloadSubscriptionProperties(int numPartitions) throws Exception {
}

// unload the topic
cluster.getService().getAdminClient().topics().unload(topicName);
pulsarContainer.getAdmin().topics().unload(topicName);

try (PulsarMessageConsumer consumer1 =
session.createSharedDurableConsumer(destination, subscriptionName, null); ) {
Expand Down Expand Up @@ -252,9 +224,8 @@ public void downloadSubscriptionProperties(int numPartitions) throws Exception {
subscriptionProperties.put("jms.selector", newSelector);
subscriptionProperties.put("jms.filtering", "true");

cluster
.getService()
.getAdminClient()
pulsarContainer
.getAdmin()
.topics()
.updateSubscriptionProperties(topicName, subscriptionName, subscriptionProperties);

Expand All @@ -275,9 +246,8 @@ public void downloadSubscriptionProperties(int numPartitions) throws Exception {
subscriptionProperties.put("jms.selector", newSelector);
subscriptionProperties.put("jms.filtering", "false");

cluster
.getService()
.getAdminClient()
pulsarContainer
.getAdmin()
.topics()
.updateSubscriptionProperties(topicName, subscriptionName, subscriptionProperties);

Expand Down
Loading
Loading