|
38 | 38 | import lombok.Cleanup; |
39 | 39 | import org.apache.bookkeeper.mledger.impl.PositionImpl; |
40 | 40 | import org.apache.http.HttpStatus; |
| 41 | +import org.apache.pulsar.broker.BrokerTestUtil; |
41 | 42 | import org.apache.pulsar.broker.ServiceConfiguration; |
42 | 43 | import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; |
43 | 44 | import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor; |
|
48 | 49 | import org.apache.pulsar.client.api.Producer; |
49 | 50 | import org.apache.pulsar.client.api.PulsarClient; |
50 | 51 | import org.apache.pulsar.client.api.Schema; |
| 52 | +import org.apache.pulsar.client.api.TransactionIsolationLevel; |
51 | 53 | import org.apache.pulsar.client.api.SubscriptionType; |
52 | 54 | import org.apache.pulsar.client.api.transaction.Transaction; |
53 | 55 | import org.apache.pulsar.client.api.transaction.TxnID; |
54 | 56 | import org.apache.pulsar.client.impl.BatchMessageIdImpl; |
55 | 57 | import org.apache.pulsar.client.impl.MessageIdImpl; |
| 58 | +import org.apache.pulsar.client.impl.MessageImpl; |
56 | 59 | import org.apache.pulsar.client.impl.transaction.TransactionImpl; |
| 60 | +import org.apache.pulsar.common.api.proto.MarkerType; |
| 61 | +import org.apache.pulsar.common.api.proto.MessageMetadata; |
57 | 62 | import org.apache.pulsar.common.naming.NamespaceName; |
58 | 63 | import org.apache.pulsar.common.naming.SystemTopicNames; |
59 | 64 | import org.apache.pulsar.common.naming.TopicDomain; |
@@ -917,6 +922,127 @@ public void testAbortTransaction() throws Exception { |
917 | 922 | } |
918 | 923 | } |
919 | 924 |
|
| 925 | + @Test |
| 926 | + public void testPeekMessageForSkipTxnMarker() throws Exception { |
| 927 | + initTransaction(1); |
| 928 | + |
| 929 | + final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/peek_marker"); |
| 930 | + |
| 931 | + @Cleanup |
| 932 | + Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); |
| 933 | + int n = 10; |
| 934 | + for (int i = 0; i < n; i++) { |
| 935 | + Transaction txn = pulsarClient.newTransaction().build().get(); |
| 936 | + producer.newMessage(txn).value("msg").send(); |
| 937 | + txn.commit().get(); |
| 938 | + } |
| 939 | + |
| 940 | + List<Message<byte[]>> peekMsgs = admin.topics().peekMessages(topic, "t-sub", n, |
| 941 | + false, TransactionIsolationLevel.READ_UNCOMMITTED); |
| 942 | + assertEquals(peekMsgs.size(), n); |
| 943 | + for (Message<byte[]> peekMsg : peekMsgs) { |
| 944 | + assertEquals(new String(peekMsg.getValue()), "msg"); |
| 945 | + } |
| 946 | + } |
| 947 | + |
| 948 | + @Test |
| 949 | + public void testPeekMessageFoReadCommittedMessages() throws Exception { |
| 950 | + initTransaction(1); |
| 951 | + |
| 952 | + final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/peek_txn"); |
| 953 | + |
| 954 | + @Cleanup |
| 955 | + Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); |
| 956 | + int n = 10; |
| 957 | + // Alternately sends `n` committed transactional messages and `n` abort transactional messages. |
| 958 | + for (int i = 0; i < 2 * n; i++) { |
| 959 | + Transaction txn = pulsarClient.newTransaction().build().get(); |
| 960 | + if (i % 2 == 0) { |
| 961 | + producer.newMessage(txn).value("msg").send(); |
| 962 | + txn.commit().get(); |
| 963 | + } else { |
| 964 | + producer.newMessage(txn).value("msg-aborted").send(); |
| 965 | + txn.abort(); |
| 966 | + } |
| 967 | + } |
| 968 | + // Then sends 1 uncommitted transactional messages. |
| 969 | + Transaction txn = pulsarClient.newTransaction().build().get(); |
| 970 | + producer.newMessage(txn).value("msg-uncommitted").send(); |
| 971 | + // Then sends n-1 no transaction messages. |
| 972 | + for (int i = 0; i < n - 1; i++) { |
| 973 | + producer.newMessage().value("msg-after-uncommitted").send(); |
| 974 | + } |
| 975 | + |
| 976 | + // peek n message, all messages value should be "msg" |
| 977 | + { |
| 978 | + List<Message<byte[]>> peekMsgs = admin.topics().peekMessages(topic, "t-sub", n, |
| 979 | + false, TransactionIsolationLevel.READ_COMMITTED); |
| 980 | + assertEquals(peekMsgs.size(), n); |
| 981 | + for (Message<byte[]> peekMsg : peekMsgs) { |
| 982 | + assertEquals(new String(peekMsg.getValue()), "msg"); |
| 983 | + } |
| 984 | + } |
| 985 | + |
| 986 | + // peek 3 * n message, and still get n message, all messages value should be "msg" |
| 987 | + { |
| 988 | + List<Message<byte[]>> peekMsgs = admin.topics().peekMessages(topic, "t-sub", 2 * n, |
| 989 | + false, TransactionIsolationLevel.READ_COMMITTED); |
| 990 | + assertEquals(peekMsgs.size(), n); |
| 991 | + for (Message<byte[]> peekMsg : peekMsgs) { |
| 992 | + assertEquals(new String(peekMsg.getValue()), "msg"); |
| 993 | + } |
| 994 | + } |
| 995 | + } |
| 996 | + |
| 997 | + @Test |
| 998 | + public void testPeekMessageForShowAllMessages() throws Exception { |
| 999 | + initTransaction(1); |
| 1000 | + |
| 1001 | + final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/peek_all"); |
| 1002 | + |
| 1003 | + @Cleanup |
| 1004 | + Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); |
| 1005 | + int n = 10; |
| 1006 | + // Alternately sends `n` committed transactional messages and `n` abort transactional messages. |
| 1007 | + for (int i = 0; i < 2 * n; i++) { |
| 1008 | + Transaction txn = pulsarClient.newTransaction().build().get(); |
| 1009 | + if (i % 2 == 0) { |
| 1010 | + producer.newMessage(txn).value("msg").send(); |
| 1011 | + txn.commit().get(); |
| 1012 | + } else { |
| 1013 | + producer.newMessage(txn).value("msg-aborted").send(); |
| 1014 | + txn.abort(); |
| 1015 | + } |
| 1016 | + } |
| 1017 | + // Then sends `n` uncommitted transactional messages. |
| 1018 | + Transaction txn = pulsarClient.newTransaction().build().get(); |
| 1019 | + for (int i = 0; i < n; i++) { |
| 1020 | + producer.newMessage(txn).value("msg-uncommitted").send(); |
| 1021 | + } |
| 1022 | + |
| 1023 | + // peek 5 * n message, will get 5 * n msg. |
| 1024 | + List<Message<byte[]>> peekMsgs = admin.topics().peekMessages(topic, "t-sub", 5 * n, |
| 1025 | + true, TransactionIsolationLevel.READ_UNCOMMITTED); |
| 1026 | + assertEquals(peekMsgs.size(), 5 * n); |
| 1027 | + |
| 1028 | + for (int i = 0; i < 4 * n; i++) { |
| 1029 | + Message<byte[]> peekMsg = peekMsgs.get(i); |
| 1030 | + MessageImpl peekMsgImpl = (MessageImpl) peekMsg; |
| 1031 | + MessageMetadata metadata = peekMsgImpl.getMessageBuilder(); |
| 1032 | + if (metadata.hasMarkerType()) { |
| 1033 | + assertTrue(metadata.getMarkerType() == MarkerType.TXN_COMMIT_VALUE || |
| 1034 | + metadata.getMarkerType() == MarkerType.TXN_ABORT_VALUE); |
| 1035 | + } else { |
| 1036 | + String value = new String(peekMsg.getValue()); |
| 1037 | + assertTrue(value.equals("msg") || value.equals("msg-aborted")); |
| 1038 | + } |
| 1039 | + } |
| 1040 | + for (int i = 4 * n; i < peekMsgs.size(); i++) { |
| 1041 | + Message<byte[]> peekMsg = peekMsgs.get(i); |
| 1042 | + assertEquals(new String(peekMsg.getValue()), "msg-uncommitted"); |
| 1043 | + } |
| 1044 | + } |
| 1045 | + |
920 | 1046 | private static void verifyCoordinatorStats(String state, |
921 | 1047 | long sequenceId, long lowWaterMark) { |
922 | 1048 | assertEquals(state, "Ready"); |
|
0 commit comments