This is an automated email from the ASF dual-hosted git repository. baodi pushed a commit to branch branch-3.3 in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.3 by this push: new 66baa8dfac9 [improve][cli] PIP-353: Improve transaction message visibility for peek-message (#22762) 66baa8dfac9 is described below commit 66baa8dfac91795cca030390d67c90a7ae43d0a8 Author: Baodi Shi <ba...@apache.org> AuthorDate: Tue May 28 12:40:52 2024 +0800 [improve][cli] PIP-353: Improve transaction message visibility for peek-message (#22762) (cherry picked from commit 20e83b96c3fcf10010977ab785093e105e4e40d8) --- .../broker/admin/impl/PersistentTopicsBase.java | 30 +++-- .../broker/admin/v3/AdminApiTransactionTest.java | 126 +++++++++++++++++++++ .../org/apache/pulsar/client/admin/Topics.java | 60 +++++++++- .../pulsar/client/admin/internal/TopicsImpl.java | 70 +++++++++--- .../client/api/TransactionIsolationLevel.java | 31 +++++ .../pulsar/admin/cli/PulsarAdminToolTest.java | 4 +- .../org/apache/pulsar/admin/cli/CmdTopics.java | 22 +++- 7 files changed, 316 insertions(+), 27 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 682f41dcdb6..3461eb2d140 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -97,6 +97,7 @@ import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedExc import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; @@ -2707,7 +2708,7 @@ public class PersistentTopicsBase extends AdminResource { @Override public void readEntryComplete(Entry entry, Object ctx) { try { - results.complete(generateResponseWithEntry(entry)); + results.complete(generateResponseWithEntry(entry, (PersistentTopic) topic)); } catch (IOException exception) { throw new RestException(exception); } finally { @@ -2848,10 +2849,12 @@ public class PersistentTopicsBase extends AdminResource { entry = sub.peekNthMessage(messagePosition); } } - return entry; - }).thenCompose(entry -> { + return entry.thenApply(e -> Pair.of(e, (PersistentTopic) topic)); + }).thenCompose(entryTopicPair -> { + Entry entry = entryTopicPair.getLeft(); + PersistentTopic persistentTopic = entryTopicPair.getRight(); try { - Response response = generateResponseWithEntry(entry); + Response response = generateResponseWithEntry(entry, persistentTopic); return CompletableFuture.completedFuture(response); } catch (NullPointerException npe) { throw new RestException(Status.NOT_FOUND, "Message not found"); @@ -2930,17 +2933,18 @@ public class PersistentTopicsBase extends AdminResource { PersistentTopicsBase.this.topicName); } }, null); - return future; + return future.thenApply(entry -> Pair.of(entry, (PersistentTopic) topic)); } catch (ManagedLedgerException exception) { log.error("[{}] Failed to examine message at position {} from {} due to {}", clientAppId(), messagePosition, topicName, exception); throw new RestException(exception); } - - }).thenApply(entry -> { + }).thenApply(entryTopicPair -> { + Entry entry = entryTopicPair.getLeft(); + PersistentTopic persistentTopic = entryTopicPair.getRight(); try { - return generateResponseWithEntry(entry); + return generateResponseWithEntry(entry, persistentTopic); } catch (IOException exception) { throw new RestException(exception); } finally { @@ -2951,7 +2955,7 @@ public class PersistentTopicsBase extends AdminResource { }); } - private Response generateResponseWithEntry(Entry entry) throws IOException { + private Response generateResponseWithEntry(Entry entry, PersistentTopic persistentTopic) throws IOException { checkNotNull(entry); PositionImpl pos = (PositionImpl) entry.getPosition(); ByteBuf metadataAndPayload = entry.getDataBuffer(); @@ -3069,6 +3073,14 @@ public class PersistentTopicsBase extends AdminResource { if (metadata.hasNullPartitionKey()) { responseBuilder.header("X-Pulsar-null-partition-key", metadata.isNullPartitionKey()); } + if (metadata.hasTxnidMostBits() && metadata.hasTxnidLeastBits()) { + TxnID txnID = new TxnID(metadata.getTxnidMostBits(), metadata.getTxnidLeastBits()); + boolean isTxnAborted = persistentTopic.isTxnAborted(txnID, (PositionImpl) entry.getPosition()); + responseBuilder.header("X-Pulsar-txn-aborted", isTxnAborted); + } + boolean isTxnUncommitted = ((PositionImpl) entry.getPosition()) + .compareTo(persistentTopic.getMaxReadPosition()) > 0; + responseBuilder.header("X-Pulsar-txn-uncommitted", isTxnUncommitted); // Decode if needed CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(metadata.getCompression()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java index adf810945de..5a192d0159a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java @@ -38,6 +38,7 @@ import java.util.concurrent.TimeUnit; import lombok.Cleanup; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.http.HttpStatus; +import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor; @@ -48,12 +49,16 @@ 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.client.api.TransactionIsolationLevel; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.api.transaction.Transaction; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.BatchMessageIdImpl; import org.apache.pulsar.client.impl.MessageIdImpl; +import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.client.impl.transaction.TransactionImpl; +import org.apache.pulsar.common.api.proto.MarkerType; +import org.apache.pulsar.common.api.proto.MessageMetadata; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.SystemTopicNames; import org.apache.pulsar.common.naming.TopicDomain; @@ -917,6 +922,127 @@ public class AdminApiTransactionTest extends MockedPulsarServiceBaseTest { } } + @Test + public void testPeekMessageForSkipTxnMarker() throws Exception { + initTransaction(1); + + final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/peek_marker"); + + @Cleanup + Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); + int n = 10; + for (int i = 0; i < n; i++) { + Transaction txn = pulsarClient.newTransaction().build().get(); + producer.newMessage(txn).value("msg").send(); + txn.commit().get(); + } + + List<Message<byte[]>> peekMsgs = admin.topics().peekMessages(topic, "t-sub", n, + false, TransactionIsolationLevel.READ_UNCOMMITTED); + assertEquals(peekMsgs.size(), n); + for (Message<byte[]> peekMsg : peekMsgs) { + assertEquals(new String(peekMsg.getValue()), "msg"); + } + } + + @Test + public void testPeekMessageFoReadCommittedMessages() throws Exception { + initTransaction(1); + + final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/peek_txn"); + + @Cleanup + Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); + int n = 10; + // Alternately sends `n` committed transactional messages and `n` abort transactional messages. + for (int i = 0; i < 2 * n; i++) { + Transaction txn = pulsarClient.newTransaction().build().get(); + if (i % 2 == 0) { + producer.newMessage(txn).value("msg").send(); + txn.commit().get(); + } else { + producer.newMessage(txn).value("msg-aborted").send(); + txn.abort(); + } + } + // Then sends 1 uncommitted transactional messages. + Transaction txn = pulsarClient.newTransaction().build().get(); + producer.newMessage(txn).value("msg-uncommitted").send(); + // Then sends n-1 no transaction messages. + for (int i = 0; i < n - 1; i++) { + producer.newMessage().value("msg-after-uncommitted").send(); + } + + // peek n message, all messages value should be "msg" + { + List<Message<byte[]>> peekMsgs = admin.topics().peekMessages(topic, "t-sub", n, + false, TransactionIsolationLevel.READ_COMMITTED); + assertEquals(peekMsgs.size(), n); + for (Message<byte[]> peekMsg : peekMsgs) { + assertEquals(new String(peekMsg.getValue()), "msg"); + } + } + + // peek 3 * n message, and still get n message, all messages value should be "msg" + { + List<Message<byte[]>> peekMsgs = admin.topics().peekMessages(topic, "t-sub", 2 * n, + false, TransactionIsolationLevel.READ_COMMITTED); + assertEquals(peekMsgs.size(), n); + for (Message<byte[]> peekMsg : peekMsgs) { + assertEquals(new String(peekMsg.getValue()), "msg"); + } + } + } + + @Test + public void testPeekMessageForShowAllMessages() throws Exception { + initTransaction(1); + + final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/peek_all"); + + @Cleanup + Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); + int n = 10; + // Alternately sends `n` committed transactional messages and `n` abort transactional messages. + for (int i = 0; i < 2 * n; i++) { + Transaction txn = pulsarClient.newTransaction().build().get(); + if (i % 2 == 0) { + producer.newMessage(txn).value("msg").send(); + txn.commit().get(); + } else { + producer.newMessage(txn).value("msg-aborted").send(); + txn.abort(); + } + } + // Then sends `n` uncommitted transactional messages. + Transaction txn = pulsarClient.newTransaction().build().get(); + for (int i = 0; i < n; i++) { + producer.newMessage(txn).value("msg-uncommitted").send(); + } + + // peek 5 * n message, will get 5 * n msg. + List<Message<byte[]>> peekMsgs = admin.topics().peekMessages(topic, "t-sub", 5 * n, + true, TransactionIsolationLevel.READ_UNCOMMITTED); + assertEquals(peekMsgs.size(), 5 * n); + + for (int i = 0; i < 4 * n; i++) { + Message<byte[]> peekMsg = peekMsgs.get(i); + MessageImpl peekMsgImpl = (MessageImpl) peekMsg; + MessageMetadata metadata = peekMsgImpl.getMessageBuilder(); + if (metadata.hasMarkerType()) { + assertTrue(metadata.getMarkerType() == MarkerType.TXN_COMMIT_VALUE || + metadata.getMarkerType() == MarkerType.TXN_ABORT_VALUE); + } else { + String value = new String(peekMsg.getValue()); + assertTrue(value.equals("msg") || value.equals("msg-aborted")); + } + } + for (int i = 4 * n; i < peekMsgs.size(); i++) { + Message<byte[]> peekMsg = peekMsgs.get(i); + assertEquals(new String(peekMsg.getValue()), "msg-uncommitted"); + } + } + private static void verifyCoordinatorStats(String state, long sequenceId, long lowWaterMark) { assertEquals(state, "Ready"); diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java index 574b859e82c..c681bd1a7bc 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java @@ -31,6 +31,7 @@ import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedExc import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.TransactionIsolationLevel; import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.AuthAction; @@ -1653,7 +1654,53 @@ public interface Topics { * @throws PulsarAdminException * Unexpected error */ - List<Message<byte[]>> peekMessages(String topic, String subName, int numMessages) throws PulsarAdminException; + default List<Message<byte[]>> peekMessages(String topic, String subName, int numMessages) + throws PulsarAdminException { + return peekMessages(topic, subName, numMessages, false, TransactionIsolationLevel.READ_COMMITTED); + } + + /** + * Peek messages from a topic subscription. + * + * @param topic + * topic name + * @param subName + * Subscription name + * @param numMessages + * Number of messages + * @param showServerMarker + * Enables the display of internal server write markers + * @param transactionIsolationLevel + * Sets the isolation level for peeking messages within transactions. + * - 'READ_COMMITTED' allows peeking only committed transactional messages. + * - 'READ_UNCOMMITTED' allows peeking all messages, + * even transactional messages which have been aborted. + * @return + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Topic or subscription does not exist + * @throws PulsarAdminException + * Unexpected error + */ + List<Message<byte[]>> peekMessages(String topic, String subName, int numMessages, + boolean showServerMarker, TransactionIsolationLevel transactionIsolationLevel) + throws PulsarAdminException; + + /** + * Peek messages from a topic subscription asynchronously. + * + * @param topic + * topic name + * @param subName + * Subscription name + * @param numMessages + * Number of messages + * @return a future that can be used to track when the messages are returned + */ + default CompletableFuture<List<Message<byte[]>>> peekMessagesAsync(String topic, String subName, int numMessages) { + return peekMessagesAsync(topic, subName, numMessages, false, TransactionIsolationLevel.READ_COMMITTED); + } /** * Peek messages from a topic subscription asynchronously. @@ -1664,9 +1711,18 @@ public interface Topics { * Subscription name * @param numMessages * Number of messages + * @param showServerMarker + * Enables the display of internal server write markers + @param transactionIsolationLevel + * Sets the isolation level for peeking messages within transactions. + * - 'READ_COMMITTED' allows peeking only committed transactional messages. + * - 'READ_UNCOMMITTED' allows peeking all messages, + * even transactional messages which have been aborted. * @return a future that can be used to track when the messages are returned */ - CompletableFuture<List<Message<byte[]>>> peekMessagesAsync(String topic, String subName, int numMessages); + CompletableFuture<List<Message<byte[]>>> peekMessagesAsync( + String topic, String subName, int numMessages, + boolean showServerMarker, TransactionIsolationLevel transactionIsolationLevel); /** * Get a message by its messageId via a topic subscription. diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java index f76cfbcde98..b7a8b876640 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java @@ -56,6 +56,7 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.TransactionIsolationLevel; import org.apache.pulsar.client.impl.BatchMessageIdImpl; import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.client.impl.MessageImpl; @@ -130,6 +131,8 @@ public class TopicsImpl extends BaseResource implements Topics { private static final String SCHEMA_VERSION = "X-Pulsar-Base64-schema-version-b64encoded"; private static final String ENCRYPTION_PARAM = "X-Pulsar-Base64-encryption-param"; private static final String ENCRYPTION_KEYS = "X-Pulsar-Base64-encryption-keys"; + public static final String TXN_ABORTED = "X-Pulsar-txn-aborted"; + public static final String TXN_UNCOMMITTED = "X-Pulsar-txn-uncommitted"; // CHECKSTYLE.ON: MemberName public static final String PROPERTY_SHADOW_SOURCE_KEY = "PULSAR.SHADOW_SOURCE"; @@ -867,7 +870,9 @@ public class TopicsImpl extends BaseResource implements Topics { return asyncPostRequest(path, Entity.entity("", MediaType.APPLICATION_JSON)); } - private CompletableFuture<List<Message<byte[]>>> peekNthMessage(String topic, String subName, int messagePosition) { + private CompletableFuture<List<Message<byte[]>>> peekNthMessage( + String topic, String subName, int messagePosition, boolean showServerMarker, + TransactionIsolationLevel transactionIsolationLevel) { TopicName tn = validateTopic(topic); String encodedSubName = Codec.encode(subName); WebTarget path = topicPath(tn, "subscription", encodedSubName, @@ -879,7 +884,8 @@ public class TopicsImpl extends BaseResource implements Topics { @Override public void completed(Response response) { try { - future.complete(getMessagesFromHttpResponse(tn.toString(), response)); + future.complete(getMessagesFromHttpResponse(tn.toString(), response, + showServerMarker, transactionIsolationLevel)); } catch (Exception e) { future.completeExceptionally(getApiException(e)); } @@ -894,28 +900,35 @@ public class TopicsImpl extends BaseResource implements Topics { } @Override - public List<Message<byte[]>> peekMessages(String topic, String subName, int numMessages) + public List<Message<byte[]>> peekMessages(String topic, String subName, int numMessages, + boolean showServerMarker, + TransactionIsolationLevel transactionIsolationLevel) throws PulsarAdminException { - return sync(() -> peekMessagesAsync(topic, subName, numMessages)); + return sync(() -> peekMessagesAsync(topic, subName, numMessages, showServerMarker, transactionIsolationLevel)); } @Override - public CompletableFuture<List<Message<byte[]>>> peekMessagesAsync(String topic, String subName, int numMessages) { + public CompletableFuture<List<Message<byte[]>>> peekMessagesAsync( + String topic, String subName, int numMessages, + boolean showServerMarker, TransactionIsolationLevel transactionIsolationLevel) { checkArgument(numMessages > 0); CompletableFuture<List<Message<byte[]>>> future = new CompletableFuture<List<Message<byte[]>>>(); - peekMessagesAsync(topic, subName, numMessages, new ArrayList<>(), future, 1); + peekMessagesAsync(topic, subName, numMessages, new ArrayList<>(), + future, 1, showServerMarker, transactionIsolationLevel); return future; } private void peekMessagesAsync(String topic, String subName, int numMessages, - List<Message<byte[]>> messages, CompletableFuture<List<Message<byte[]>>> future, int nthMessage) { + List<Message<byte[]>> messages, CompletableFuture<List<Message<byte[]>>> future, int nthMessage, + boolean showServerMarker, TransactionIsolationLevel transactionIsolationLevel) { if (numMessages <= 0) { future.complete(messages); return; } // if peeking first message succeeds, we know that the topic and subscription exists - peekNthMessage(topic, subName, nthMessage).handle((r, ex) -> { + peekNthMessage(topic, subName, nthMessage, showServerMarker, transactionIsolationLevel) + .handle((r, ex) -> { if (ex != null) { // if we get a not found exception, it means that the position for the message we are trying to get // does not exist. At this point, we can return the already found messages. @@ -930,7 +943,8 @@ public class TopicsImpl extends BaseResource implements Topics { for (int i = 0; i < Math.min(r.size(), numMessages); i++) { messages.add(r.get(i)); } - peekMessagesAsync(topic, subName, numMessages - r.size(), messages, future, nthMessage + 1); + peekMessagesAsync(topic, subName, numMessages - r.size(), messages, future, + nthMessage + 1, showServerMarker, transactionIsolationLevel); return null; }); } @@ -1253,6 +1267,13 @@ public class TopicsImpl extends BaseResource implements Topics { } private List<Message<byte[]>> getMessagesFromHttpResponse(String topic, Response response) throws Exception { + return getMessagesFromHttpResponse(topic, response, true, + TransactionIsolationLevel.READ_UNCOMMITTED); + } + + private List<Message<byte[]>> getMessagesFromHttpResponse( + String topic, Response response, boolean showServerMarker, + TransactionIsolationLevel transactionIsolationLevel) throws Exception { if (response.getStatus() != Status.OK.getStatusCode()) { throw getApiException(response); @@ -1284,7 +1305,32 @@ public class TopicsImpl extends BaseResource implements Topics { Map<String, String> properties = new TreeMap<>(); MultivaluedMap<String, Object> headers = response.getHeaders(); - Object tmp = headers.getFirst(PUBLISH_TIME); + Object tmp = headers.getFirst(MARKER_TYPE); + if (tmp != null) { + if (!showServerMarker) { + return new ArrayList<>(); + } else { + messageMetadata.setMarkerType(Integer.parseInt(tmp.toString())); + } + } + + tmp = headers.getFirst(TXN_ABORTED); + if (tmp != null && Boolean.parseBoolean(tmp.toString())) { + properties.put(TXN_ABORTED, tmp.toString()); + if (transactionIsolationLevel == TransactionIsolationLevel.READ_COMMITTED) { + return new ArrayList<>(); + } + } + + tmp = headers.getFirst(TXN_UNCOMMITTED); + if (tmp != null && Boolean.parseBoolean(tmp.toString())) { + properties.put(TXN_UNCOMMITTED, tmp.toString()); + if (transactionIsolationLevel == TransactionIsolationLevel.READ_COMMITTED) { + return new ArrayList<>(); + } + } + + tmp = headers.getFirst(PUBLISH_TIME); if (tmp != null) { messageMetadata.setPublishTime(DateFormatter.parse(tmp.toString())); } @@ -1336,10 +1382,6 @@ public class TopicsImpl extends BaseResource implements Topics { if (tmp != null) { messageMetadata.setPartitionKeyB64Encoded(Boolean.parseBoolean(tmp.toString())); } - tmp = headers.getFirst(MARKER_TYPE); - if (tmp != null) { - messageMetadata.setMarkerType(Integer.parseInt(tmp.toString())); - } tmp = headers.getFirst(TXNID_LEAST_BITS); if (tmp != null) { messageMetadata.setTxnidLeastBits(Long.parseLong(tmp.toString())); diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TransactionIsolationLevel.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TransactionIsolationLevel.java new file mode 100644 index 00000000000..ae385b20232 --- /dev/null +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TransactionIsolationLevel.java @@ -0,0 +1,31 @@ +/* + * 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.client.api; + +import org.apache.pulsar.common.classification.InterfaceAudience; +import org.apache.pulsar.common.classification.InterfaceStability; + +@InterfaceAudience.Public +@InterfaceStability.Stable +public enum TransactionIsolationLevel { + // Consumer can only consume all transactional messages which have been committed. + READ_COMMITTED, + // Consumer can consume all messages, even transactional messages which have been aborted. + READ_UNCOMMITTED; +} diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java index fd1bdf47998..a3b1fa075cf 100644 --- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java +++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java @@ -81,6 +81,7 @@ import org.apache.pulsar.client.admin.Transactions; import org.apache.pulsar.client.admin.internal.OffloadProcessStatusImpl; import org.apache.pulsar.client.admin.internal.PulsarAdminImpl; import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.TransactionIsolationLevel; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.api.transaction.TxnID; @@ -1744,7 +1745,8 @@ public class PulsarAdminToolTest { verify(mockTopics).deletePartitionedTopic("persistent://myprop/clust/ns1/ds1", true); cmdTopics.run(split("peek-messages persistent://myprop/clust/ns1/ds1 -s sub1 -n 3")); - verify(mockTopics).peekMessages("persistent://myprop/clust/ns1/ds1", "sub1", 3); + verify(mockTopics).peekMessages("persistent://myprop/clust/ns1/ds1", "sub1", 3, + false, TransactionIsolationLevel.READ_COMMITTED); MessageImpl message = mock(MessageImpl.class); when(message.getData()).thenReturn(new byte[]{}); diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java index e1e85c68f7e..261bd81a5b7 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java @@ -62,9 +62,12 @@ import org.apache.pulsar.client.admin.Topics; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.TransactionIsolationLevel; import org.apache.pulsar.client.impl.BatchMessageIdImpl; import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.client.impl.MessageImpl; +import org.apache.pulsar.common.api.proto.MarkerType; +import org.apache.pulsar.common.api.proto.MessageMetadata; import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.BacklogQuota; @@ -1097,10 +1100,23 @@ public class CmdTopics extends CmdBase { @Option(names = { "-n", "--count" }, description = "Number of messages (default 1)", required = false) private int numMessages = 1; + @Option(names = { "-ssm", "--show-server-marker" }, + description = "Enables the display of internal server write markers.", required = false) + private boolean showServerMarker = false; + + @Option(names = { "-til", "--transaction-isolation-level" }, + description = "Sets the isolation level for peeking messages within transactions. " + + "'READ_COMMITTED' allows peeking only committed transactional messages. " + + "'READ_UNCOMMITTED' allows peeking all messages, " + + "even transactional messages which have been aborted.", + required = false) + private TransactionIsolationLevel transactionIsolationLevel = TransactionIsolationLevel.READ_COMMITTED; + @Override void run() throws PulsarAdminException { String persistentTopic = validatePersistentTopic(topicName); - List<Message<byte[]>> messages = getTopics().peekMessages(persistentTopic, subName, numMessages); + List<Message<byte[]>> messages = getTopics().peekMessages(persistentTopic, subName, numMessages, + showServerMarker, transactionIsolationLevel); int position = 0; for (Message<byte[]> msg : messages) { MessageImpl message = (MessageImpl) msg; @@ -1122,6 +1138,10 @@ public class CmdTopics extends CmdBase { if (message.getDeliverAtTime() != 0) { System.out.println("Deliver at time: " + message.getDeliverAtTime()); } + MessageMetadata msgMetaData = message.getMessageBuilder(); + if (showServerMarker && msgMetaData.hasMarkerType()) { + System.out.println("Marker Type: " + MarkerType.valueOf(msgMetaData.getMarkerType())); + } if (message.getBrokerEntryMetadata() != null) { if (message.getBrokerEntryMetadata().hasBrokerTimestamp()) {