From f7878c4f9a398f287fe49119af137f126b52f5d2 Mon Sep 17 00:00:00 2001 From: lipenghui Date: Wed, 13 Nov 2019 22:27:09 +0800 Subject: [PATCH 1/5] Add highest sequenceId for CommandSendReceipt. --- .../pulsar/broker/service/Producer.java | 3 +- .../pulsar/broker/service/ServerCnx.java | 3 +- .../pulsar/client/api/ClientErrorsTest.java | 4 +- .../pulsar/client/api/MockBrokerService.java | 2 +- .../apache/pulsar/client/impl/ClientCnx.java | 3 +- .../pulsar/client/impl/ProducerImpl.java | 46 ++++++++------- .../pulsar/common/api/proto/PulsarApi.java | 57 +++++++++++++++++++ .../pulsar/common/protocol/Commands.java | 4 +- pulsar-common/src/main/proto/PulsarApi.proto | 1 + 9 files changed, 95 insertions(+), 28 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 5ead8c944ad77..fed6a6f3ea91a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -377,9 +377,8 @@ public void run() { // stats rateIn.recordMultipleEvents(batchSize, msgSize); producer.topic.recordAddLatency(System.nanoTime() - startTimeNs, TimeUnit.NANOSECONDS); - long callBackSequenceId = Math.max(highestSequenceId, sequenceId); producer.cnx.ctx().writeAndFlush( - Commands.newSendReceipt(producer.producerId, callBackSequenceId, ledgerId, entryId), + Commands.newSendReceipt(producer.producerId, sequenceId, highestSequenceId, ledgerId, entryId), producer.cnx.ctx().voidPromise()); producer.cnx.completedSendOperation(producer.isNonPersistentTopic); producer.publishOperationCompleted(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 6b38d377b9773..ce138b352b4e2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -1038,8 +1038,9 @@ protected void handleSend(CommandSend send, ByteBuf headersAndPayload) { if (nonPersistentPendingMessages > MaxNonPersistentPendingMessages) { final long producerId = send.getProducerId(); final long sequenceId = send.getSequenceId(); + final long highestSequenceId = send.getHighestSequenceId(); service.getTopicOrderedExecutor().executeOrdered(producer.getTopic().getName(), SafeRun.safeRun(() -> { - ctx.writeAndFlush(Commands.newSendReceipt(producerId, sequenceId, -1, -1), ctx.voidPromise()); + ctx.writeAndFlush(Commands.newSendReceipt(producerId, sequenceId, highestSequenceId, -1, -1), ctx.voidPromise()); })); producer.recordMessageDrop(send.getNumMessages()); return; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java index f8ac75b819ec0..c07b896f5947d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java @@ -267,7 +267,7 @@ private void producerContinuousRetryAfterSendFail(String topic) throws Exception ctx.writeAndFlush(Commands.newSendError(0, 0, ServerError.PersistenceError, "Send Failed")); return; } - ctx.writeAndFlush(Commands.newSendReceipt(0, 0, 1, 1)); + ctx.writeAndFlush(Commands.newSendReceipt(0, 0, 0, 1, 1)); }); try { @@ -587,7 +587,7 @@ public void testProducerReconnect() throws Exception { mockBrokerService.setHandleSend((ctx, sendCmd, headersAndPayload) -> { msgSent.set(true); - ctx.writeAndFlush(Commands.newSendReceipt(0, 0, 1, 1)); + ctx.writeAndFlush(Commands.newSendReceipt(0, 0, 0, 1, 1)); }); PulsarClient client = PulsarClient.builder().serviceUrl("http://127.0.0.1:" + WEB_SERVICE_PORT).build(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java index 11462452dbe3d..134c5715e7aa9 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java @@ -185,7 +185,7 @@ protected void handleSend(CommandSend send, ByteBuf headersAndPayload) { return; } // default - ctx.writeAndFlush(Commands.newSendReceipt(producerId, send.getSequenceId(), 0, 0)); + ctx.writeAndFlush(Commands.newSendReceipt(producerId, send.getSequenceId(), 0, 0, 0)); } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java index 4025ca3031200..a7abeaaccdb0f 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java @@ -347,6 +347,7 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) { long producerId = sendReceipt.getProducerId(); long sequenceId = sendReceipt.getSequenceId(); + long highestSequenceId = sendReceipt.getHighestSequenceId(); long ledgerId = -1; long entryId = -1; if (sendReceipt.hasMessageId()) { @@ -364,7 +365,7 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) { ledgerId, entryId); } - producers.get(producerId).ackReceived(this, sequenceId, ledgerId, entryId); + producers.get(producerId).ackReceived(this, sequenceId, highestSequenceId, ledgerId, entryId); } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java index 2ccbfe19c50ad..76d7753da8832 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java @@ -788,38 +788,46 @@ public void terminated(ClientCnx cnx) { } } - void ackReceived(ClientCnx cnx, long sequenceId, long ledgerId, long entryId) { + void ackReceived(ClientCnx cnx, long sequenceId, long highestSequenceId, long ledgerId, long entryId) { OpSendMsg op = null; boolean callback = false; synchronized (this) { op = pendingMessages.peek(); if (op == null) { if (log.isDebugEnabled()) { - log.debug("[{}] [{}] Got ack for timed out msg {}", topic, producerName, sequenceId); + log.debug("[{}] [{}] Got ack for timed out msg {} - {}", topic, producerName, sequenceId, highestSequenceId); } return; } - long expectedSequenceId = getHighestSequenceId(op); - if (sequenceId > expectedSequenceId) { - log.warn("[{}] [{}] Got ack for msg. expecting: {} - got: {} - queue-size: {}", topic, producerName, - expectedSequenceId, sequenceId, pendingMessages.size()); + + if (sequenceId > op.sequenceId) { + log.warn("[{}] [{}] Got ack for msg. expecting: {} - {} - got: {} - {} - queue-size: {}", topic, producerName, + op.sequenceId, op.highestSequenceId, sequenceId, highestSequenceId, pendingMessages.size()); // Force connection closing so that messages can be re-transmitted in a new connection cnx.channel().close(); - } else if (sequenceId < expectedSequenceId) { + } else if (sequenceId < op.sequenceId) { // Ignoring the ack since it's referring to a message that has already timed out. if (log.isDebugEnabled()) { - log.debug("[{}] [{}] Got ack for timed out msg {} last-seq: {}", topic, producerName, sequenceId, - expectedSequenceId); + log.debug("[{}] [{}] Got ack for timed out msg. expecting: {} - {} - got: {} - {}", topic, producerName, + op.sequenceId, op.highestSequenceId, sequenceId, highestSequenceId); } } else { - // Message was persisted correctly - if (log.isDebugEnabled()) { - log.debug("[{}] [{}] Received ack for msg {} ", topic, producerName, sequenceId); + boolean persistentSuccess = sequenceId >= highestSequenceId || highestSequenceId == op.highestSequenceId; + if (persistentSuccess) { + // Message was persisted correctly + if (log.isDebugEnabled()) { + log.debug("[{}] [{}] Received ack for msg {} ", topic, producerName, sequenceId); + } + pendingMessages.remove(); + releaseSemaphoreForSendOp(op); + callback = true; + pendingCallbacks.add(op); + } else { + log.warn("[{}] [{}] Got ack for batch msg error. expecting: {} - {} - got: {} - {} - queue-size: {}", topic, producerName, + op.sequenceId, op.highestSequenceId, sequenceId, highestSequenceId, pendingMessages.size()); + // Force connection closing so that messages can be re-transmitted in a new connection + cnx.channel().close(); } - pendingMessages.remove(); - releaseSemaphoreForSendOp(op); - callback = true; - pendingCallbacks.add(op); } } if (callback) { @@ -844,6 +852,7 @@ void ackReceived(ClientCnx cnx, long sequenceId, long ledgerId, long entryId) { private long getHighestSequenceId(OpSendMsg op) { return Math.max(op.highestSequenceId, op.sequenceId); } + private void releaseSemaphoreForSendOp(OpSendMsg op) { semaphore.release(isBatchMessagingEnabled() ? op.numMessagesInBatch : 1); } @@ -952,7 +961,6 @@ protected static final class OpSendMsg { long createdAt; long batchSizeByte = 0; int numMessagesInBatch = 1; - long lowestSequenceId; long highestSequenceId; static OpSendMsg create(MessageImpl msg, ByteBufPair cmd, long sequenceId, SendCallback callback) { @@ -981,9 +989,8 @@ static OpSendMsg create(List> msgs, ByteBufPair cmd, long lowestS op.msgs = msgs; op.cmd = cmd; op.callback = callback; - op.lowestSequenceId = lowestSequenceId; - op.highestSequenceId = highestSequenceId; op.sequenceId = lowestSequenceId; + op.highestSequenceId = highestSequenceId; op.createdAt = System.currentTimeMillis(); return op; } @@ -996,7 +1003,6 @@ void recycle() { rePopulate = null; sequenceId = -1L; createdAt = -1L; - lowestSequenceId = -1L; highestSequenceId = -1L; recyclerHandle.recycle(this); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java b/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java index dcda3df6dbe44..2b477e9add568 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java @@ -15997,6 +15997,10 @@ public interface CommandSendReceiptOrBuilder // optional .pulsar.proto.MessageIdData message_id = 3; boolean hasMessageId(); org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData getMessageId(); + + // optional uint64 highest_sequence_id = 4 [default = 0]; + boolean hasHighestSequenceId(); + long getHighestSequenceId(); } public static final class CommandSendReceipt extends org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite @@ -16063,10 +16067,21 @@ public org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData getMessageId() return messageId_; } + // optional uint64 highest_sequence_id = 4 [default = 0]; + public static final int HIGHEST_SEQUENCE_ID_FIELD_NUMBER = 4; + private long highestSequenceId_; + public boolean hasHighestSequenceId() { + return ((bitField0_ & 0x00000008) == 0x00000008); + } + public long getHighestSequenceId() { + return highestSequenceId_; + } + private void initFields() { producerId_ = 0L; sequenceId_ = 0L; messageId_ = org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData.getDefaultInstance(); + highestSequenceId_ = 0L; } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { @@ -16108,6 +16123,9 @@ public void writeTo(org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStr if (((bitField0_ & 0x00000004) == 0x00000004)) { output.writeMessage(3, messageId_); } + if (((bitField0_ & 0x00000008) == 0x00000008)) { + output.writeUInt64(4, highestSequenceId_); + } } private int memoizedSerializedSize = -1; @@ -16128,6 +16146,10 @@ public int getSerializedSize() { size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream .computeMessageSize(3, messageId_); } + if (((bitField0_ & 0x00000008) == 0x00000008)) { + size += org.apache.pulsar.shaded.com.google.protobuf.v241.CodedOutputStream + .computeUInt64Size(4, highestSequenceId_); + } memoizedSerializedSize = size; return size; } @@ -16247,6 +16269,8 @@ public Builder clear() { bitField0_ = (bitField0_ & ~0x00000002); messageId_ = org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData.getDefaultInstance(); bitField0_ = (bitField0_ & ~0x00000004); + highestSequenceId_ = 0L; + bitField0_ = (bitField0_ & ~0x00000008); return this; } @@ -16292,6 +16316,10 @@ public org.apache.pulsar.common.api.proto.PulsarApi.CommandSendReceipt buildPart to_bitField0_ |= 0x00000004; } result.messageId_ = messageId_; + if (((from_bitField0_ & 0x00000008) == 0x00000008)) { + to_bitField0_ |= 0x00000008; + } + result.highestSequenceId_ = highestSequenceId_; result.bitField0_ = to_bitField0_; return result; } @@ -16307,6 +16335,9 @@ public Builder mergeFrom(org.apache.pulsar.common.api.proto.PulsarApi.CommandSen if (other.hasMessageId()) { mergeMessageId(other.getMessageId()); } + if (other.hasHighestSequenceId()) { + setHighestSequenceId(other.getHighestSequenceId()); + } return this; } @@ -16370,6 +16401,11 @@ public Builder mergeFrom( subBuilder.recycle(); break; } + case 32: { + bitField0_ |= 0x00000008; + highestSequenceId_ = input.readUInt64(); + break; + } } } } @@ -16461,6 +16497,27 @@ public Builder clearMessageId() { return this; } + // optional uint64 highest_sequence_id = 4 [default = 0]; + private long highestSequenceId_ ; + public boolean hasHighestSequenceId() { + return ((bitField0_ & 0x00000008) == 0x00000008); + } + public long getHighestSequenceId() { + return highestSequenceId_; + } + public Builder setHighestSequenceId(long value) { + bitField0_ |= 0x00000008; + highestSequenceId_ = value; + + return this; + } + public Builder clearHighestSequenceId() { + bitField0_ = (bitField0_ & ~0x00000008); + highestSequenceId_ = 0L; + + return this; + } + // @@protoc_insertion_point(builder_scope:pulsar.proto.CommandSendReceipt) } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java index 4322b3a94ad59..c067a8981b587 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java @@ -336,10 +336,12 @@ public static ByteBuf newError(long requestId, ServerError error, String message } - public static ByteBuf newSendReceipt(long producerId, long sequenceId, long ledgerId, long entryId) { + public static ByteBuf newSendReceipt(long producerId, long sequenceId, long highestId, long ledgerId, + long entryId) { CommandSendReceipt.Builder sendReceiptBuilder = CommandSendReceipt.newBuilder(); sendReceiptBuilder.setProducerId(producerId); sendReceiptBuilder.setSequenceId(sequenceId); + sendReceiptBuilder.setHighestSequenceId(highestId); MessageIdData.Builder messageIdBuilder = MessageIdData.newBuilder(); messageIdBuilder.setLedgerId(ledgerId); messageIdBuilder.setEntryId(entryId); diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index f9c88b5f8a09f..c20a382c637de 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -423,6 +423,7 @@ message CommandSendReceipt { required uint64 producer_id = 1; required uint64 sequence_id = 2; optional MessageIdData message_id = 3; + optional uint64 highest_sequence_id = 4 [default = 0]; } message CommandSendError { From 9e76a39eb7834f586253da9bb6fdd0de1e9067e1 Mon Sep 17 00:00:00 2001 From: lipenghui Date: Thu, 14 Nov 2019 11:13:10 +0800 Subject: [PATCH 2/5] Add backward compatibility tests for batch message publish. --- .../PulsarStandaloneTestSuite2_3.java | 42 +++++++++++++++++++ .../PulsarStandaloneTestSuite2_4.java | 42 +++++++++++++++++++ .../backwardscompatibility/SmokeTest2_2.java | 5 +++ .../backwardscompatibility/SmokeTest2_3.java | 36 ++++++++++++++++ .../backwardscompatibility/SmokeTest2_4.java | 36 ++++++++++++++++ .../containers/PulsarContainer.java | 1 + .../topologies/PulsarTestBase.java | 37 ++++++++++++++++ .../pulsar-backwards-compatibility.xml | 30 +++++++++++++ .../integration/src/test/resources/pulsar.xml | 1 + 9 files changed, 230 insertions(+) create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_3.java create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_4.java create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_3.java create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_4.java create mode 100644 tests/integration/src/test/resources/pulsar-backwards-compatibility.xml diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_3.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_3.java new file mode 100644 index 0000000000000..7f0539b7a8163 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_3.java @@ -0,0 +1,42 @@ +/** + * 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.tests.integration.backwardscompatibility; + +import org.apache.pulsar.tests.integration.containers.PulsarContainer; +import org.apache.pulsar.tests.integration.topologies.PulsarStandaloneTestBase; +import org.testng.ITest; +import org.testng.annotations.AfterSuite; +import org.testng.annotations.BeforeSuite; + +public class PulsarStandaloneTestSuite2_3 extends PulsarStandaloneTestBase implements ITest { + + @BeforeSuite + public void setUpCluster() throws Exception { + super.startCluster(PulsarContainer.PULSAR_2_3_IMAGE_NAME); + } + + @AfterSuite + public void tearDownCluster() throws Exception { + super.stopCluster(); + } + @Override + public String getTestName() { + return "pulsar-standalone-suite"; + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_4.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_4.java new file mode 100644 index 0000000000000..08c3e2b93c644 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/PulsarStandaloneTestSuite2_4.java @@ -0,0 +1,42 @@ +/** + * 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.tests.integration.backwardscompatibility; + +import org.apache.pulsar.tests.integration.containers.PulsarContainer; +import org.apache.pulsar.tests.integration.topologies.PulsarStandaloneTestBase; +import org.testng.ITest; +import org.testng.annotations.AfterSuite; +import org.testng.annotations.BeforeSuite; + +public class PulsarStandaloneTestSuite2_4 extends PulsarStandaloneTestBase implements ITest { + + @BeforeSuite + public void setUpCluster() throws Exception { + super.startCluster(PulsarContainer.PULSAR_2_4_IMAGE_NAME); + } + + @AfterSuite + public void tearDownCluster() throws Exception { + super.stopCluster(); + } + @Override + public String getTestName() { + return "pulsar-standalone-suite"; + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_2.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_2.java index d9c446d3bcc56..7c1c2a19b0313 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_2.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_2.java @@ -28,4 +28,9 @@ public void testPublishAndConsume(String serviceUrl, boolean isPersistent) throw super.testPublishAndConsume(serviceUrl, isPersistent); } + @Test(dataProvider = "StandaloneServiceUrlAndTopics") + public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersistent) throws Exception { + super.testBatchMessagePublishAndConsume(serviceUrl, isPersistent); + } + } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_3.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_3.java new file mode 100644 index 0000000000000..ab317d043d00f --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_3.java @@ -0,0 +1,36 @@ +/** + * 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.tests.integration.backwardscompatibility; + +import org.testng.annotations.Test; + +public class SmokeTest2_3 extends PulsarStandaloneTestSuite2_3 { + + @Test(dataProvider = "StandaloneServiceUrlAndTopics") + public void testPublishAndConsume(String serviceUrl, boolean isPersistent) throws Exception { + super.testPublishAndConsume(serviceUrl, isPersistent); + } + + @Test(dataProvider = "StandaloneServiceUrlAndTopics") + public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersistent) throws Exception { + super.testBatchMessagePublishAndConsume(serviceUrl, isPersistent); + } + +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_4.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_4.java new file mode 100644 index 0000000000000..d74ad8eecff1d --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/SmokeTest2_4.java @@ -0,0 +1,36 @@ +/** + * 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.tests.integration.backwardscompatibility; + +import org.testng.annotations.Test; + +public class SmokeTest2_4 extends PulsarStandaloneTestSuite2_4 { + + @Test(dataProvider = "StandaloneServiceUrlAndTopics") + public void testPublishAndConsume(String serviceUrl, boolean isPersistent) throws Exception { + super.testPublishAndConsume(serviceUrl, isPersistent); + } + + @Test(dataProvider = "StandaloneServiceUrlAndTopics") + public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersistent) throws Exception { + super.testBatchMessagePublishAndConsume(serviceUrl, isPersistent); + } + +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java index eff01e4622976..15c99cc9e1868 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java @@ -41,6 +41,7 @@ public abstract class PulsarContainer> exte public static final int BROKER_HTTP_PORT = 8080; public static final String DEFAULT_IMAGE_NAME = "apachepulsar/pulsar-test-latest-version:latest"; + public static final String PULSAR_2_4_IMAGE_NAME = "apachepulsar/pulsar:2.4.0"; public static final String PULSAR_2_3_IMAGE_NAME = "apachepulsar/pulsar:2.3.0"; public static final String PULSAR_2_2_IMAGE_NAME = "apachepulsar/pulsar:2.2.0"; public static final String PULSAR_2_1_IMAGE_NAME = "apachepulsar/pulsar:2.1.0"; diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java index 2724be2003aee..a5f64b3abe204 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java @@ -20,12 +20,17 @@ import static org.testng.Assert.assertEquals; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ThreadLocalRandom; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.util.FutureUtil; public class PulsarTestBase { @@ -90,4 +95,36 @@ public void testPublishAndConsume(String serviceUrl, boolean isPersistent) throw } } + public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersistent) throws Exception { + String topicName = generateTopicName("test-batch-publish-consume", isPersistent); + + final int numMessages = 10000; + try (PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build()) { + try (Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("my-sub") + .subscribe()) { + + try (Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .create()) { + + List> futures = new ArrayList<>(); + for (int i = 0; i < numMessages; i++) { + futures.add(producer.sendAsync("smoke-message-" + i)); + } + // Wait for all messages are publish succeed. + FutureUtil.waitForAll(futures).get(); + } + + for (int i = 0; i < numMessages; i++) { + Message m = consumer.receive(); + assertEquals("smoke-message-" + i, m.getValue()); + } + } + } + } + } diff --git a/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml b/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml new file mode 100644 index 0000000000000..193c7e220b227 --- /dev/null +++ b/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/tests/integration/src/test/resources/pulsar.xml b/tests/integration/src/test/resources/pulsar.xml index e9339c43f760f..44a46fadb30a8 100644 --- a/tests/integration/src/test/resources/pulsar.xml +++ b/tests/integration/src/test/resources/pulsar.xml @@ -32,5 +32,6 @@ + From f38355b0840502678a6aec7cdbd6f0b5e9022e5f Mon Sep 17 00:00:00 2001 From: lipenghui Date: Thu, 14 Nov 2019 11:20:10 +0800 Subject: [PATCH 3/5] Fix typo and add more comments --- .../main/java/org/apache/pulsar/client/impl/ProducerImpl.java | 4 ++-- .../src/test/resources/pulsar-backwards-compatibility.xml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java index 76d7753da8832..aa8dd7e3ee010 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java @@ -812,8 +812,8 @@ void ackReceived(ClientCnx cnx, long sequenceId, long highestSequenceId, long le op.sequenceId, op.highestSequenceId, sequenceId, highestSequenceId); } } else { - boolean persistentSuccess = sequenceId >= highestSequenceId || highestSequenceId == op.highestSequenceId; - if (persistentSuccess) { + // Add check `sequenceId >= highestSequenceId` for backward compatibility. + if (sequenceId >= highestSequenceId || highestSequenceId == op.highestSequenceId) { // Message was persisted correctly if (log.isDebugEnabled()) { log.debug("[{}] [{}] Received ack for msg {} ", topic, producerName, sequenceId); diff --git a/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml b/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml index 193c7e220b227..b566ca29f6ed8 100644 --- a/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml +++ b/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml @@ -19,7 +19,7 @@ --> - + From f860df636dc55361d1b796178f17832148ea7a4a Mon Sep 17 00:00:00 2001 From: Penghui Li Date: Thu, 14 Nov 2019 14:08:27 +0800 Subject: [PATCH 4/5] Fix integration tests issue. --- .../integration/topologies/PulsarTestBase.java | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java index a5f64b3abe204..dbd3c4796f5ff 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java @@ -100,16 +100,18 @@ public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersi final int numMessages = 10000; try (PulsarClient client = PulsarClient.builder() - .serviceUrl(serviceUrl) - .build()) { + .serviceUrl(serviceUrl) + .build()) { + try (Consumer consumer = client.newConsumer(Schema.STRING) - .topic(topicName) - .subscriptionName("my-sub") - .subscribe()) { + .topic(topicName) + .subscriptionName("my-sub") + .subscribe()) { try (Producer producer = client.newProducer(Schema.STRING) - .topic(topicName) - .create()) { + .topic(topicName) + .blockIfQueueFull(true) + .create()) { List> futures = new ArrayList<>(); for (int i = 0; i < numMessages; i++) { From 1331e43861e0773b34485fb14512894c21bed6fb Mon Sep 17 00:00:00 2001 From: lipenghui Date: Thu, 14 Nov 2019 19:05:11 +0800 Subject: [PATCH 5/5] Fix non-persistent topic batch messages test issue. --- .../pulsar/tests/integration/topologies/PulsarTestBase.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java index dbd3c4796f5ff..589a62a44e67d 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarTestBase.java @@ -105,6 +105,7 @@ public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersi try (Consumer consumer = client.newConsumer(Schema.STRING) .topic(topicName) + .receiverQueueSize(10000) .subscriptionName("my-sub") .subscribe()) { @@ -123,6 +124,7 @@ public void testBatchMessagePublishAndConsume(String serviceUrl, boolean isPersi for (int i = 0; i < numMessages; i++) { Message m = consumer.receive(); + System.out.println(i); assertEquals("smoke-message-" + i, m.getValue()); } }