diff --git a/amqp-client-auth/pom.xml b/amqp-client-auth/pom.xml index 41165c44..dfd64228 100644 --- a/amqp-client-auth/pom.xml +++ b/amqp-client-auth/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-amqp-parent io.streamnative.pulsar.handlers - 3.3.0-SNAPSHOT + 4.2.1.6 4.0.0 diff --git a/amqp-impl/pom.xml b/amqp-impl/pom.xml index b28e6529..8eaa3a32 100644 --- a/amqp-impl/pom.xml +++ b/amqp-impl/pom.xml @@ -22,7 +22,7 @@ io.streamnative.pulsar.handlers pulsar-protocol-handler-amqp-parent - 3.3.0-SNAPSHOT + 4.2.1.6 pulsar-protocol-handler-amqp diff --git a/pom.xml b/pom.xml index f3bfc1f2..7e6b9691 100644 --- a/pom.xml +++ b/pom.xml @@ -29,7 +29,7 @@ io.streamnative.pulsar.handlers pulsar-protocol-handler-amqp-parent - 3.3.0-SNAPSHOT + 4.2.1.6 StreamNative :: Pulsar Protocol Handler :: AoP Parent Parent for AMQP on Pulsar implemented using Pulsar Protocol Handler. @@ -40,10 +40,10 @@ ${maven.compiler.target} - 4.2.0-SNAPSHOT + 4.2.1.6 8.0.0 5.8.0 - 4.2.0-SNAPSHOT + 4.2.1.6 1.56.0 diff --git a/tests-qpid-jms-client/pom.xml b/tests-qpid-jms-client/pom.xml index 1e95ec8e..b7a20354 100644 --- a/tests-qpid-jms-client/pom.xml +++ b/tests-qpid-jms-client/pom.xml @@ -20,7 +20,7 @@ io.streamnative.pulsar.handlers pulsar-protocol-handler-amqp-parent - 3.3.0-SNAPSHOT + 4.2.1.6 4.0.0 diff --git a/tests/pom.xml b/tests/pom.xml index 889c5050..3f7db91d 100644 --- a/tests/pom.xml +++ b/tests/pom.xml @@ -22,7 +22,7 @@ io.streamnative.pulsar.handlers pulsar-protocol-handler-amqp-parent - 3.3.0-SNAPSHOT + 4.2.1.6 io.streamnative.pulsar.handlers diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/amqp/MultiBundlesTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/amqp/MultiBundlesTest.java index df7c04be..eefa379d 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/amqp/MultiBundlesTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/amqp/MultiBundlesTest.java @@ -74,7 +74,7 @@ public void e2eTest() throws Exception { AtomicInteger receiveCount = new AtomicInteger(); AtomicBoolean flag1 = new AtomicBoolean(false); - channel.basicConsume(qu1, false, new DefaultConsumer(channel) { + channel.basicConsume(qu1, true, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { @@ -82,12 +82,11 @@ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProp if (!new String(body).contains(key1)) { flag1.set(true); } - channel.basicAck(envelope.getDeliveryTag(), false); } }); AtomicBoolean flag2 = new AtomicBoolean(false); - channel.basicConsume(qu2, false, new DefaultConsumer(channel) { + channel.basicConsume(qu2, true, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { @@ -95,7 +94,6 @@ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProp if (!new String(body).contains(key2)) { flag2.set(true); } - channel.basicAck(envelope.getDeliveryTag(), false); } });