You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/07/30 06:14:42 UTC
[pulsar] 01/02: Enable peeking encrypted batch messages (#11244)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d3bc1eb9158f5f218d33e028fd5b6882928fc73e
Author: Masahiro Sakamoto <ma...@yahoo-corp.jp>
AuthorDate: Thu Jul 8 05:54:11 2021 +0900
Enable peeking encrypted batch messages (#11244)
(cherry picked from commit d01ecb041847070eb1d011ebccadab5ee1d40ca3)
---
.../broker/admin/impl/PersistentTopicsBase.java | 1 +
.../apache/pulsar/broker/admin/AdminApiTest.java | 43 ++++++++++++++++++++++
.../pulsar/client/admin/internal/TopicsImpl.java | 11 +++++-
3 files changed, 54 insertions(+), 1 deletion(-)
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 781de11..234f153 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
@@ -2503,6 +2503,7 @@ public class PersistentTopicsBase extends AdminResource {
responseBuilder.header("X-Pulsar-PROPERTY-TOTAL-CHUNKS", Integer.toString(metadata.getNumChunksFromMsg()));
responseBuilder.header("X-Pulsar-PROPERTY-CHUNK-ID", Integer.toString(metadata.getChunkId()));
}
+ responseBuilder.header("X-Pulsar-Is-Encrypted", metadata.getEncryptionKeysCount() > 0);
// Decode if needed
CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(metadata.getCompression());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index fb8f799..82d7b1c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -84,6 +84,7 @@ import org.apache.pulsar.client.admin.internal.TenantsImpl;
import org.apache.pulsar.client.admin.internal.TopicsImpl;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRoutingMode;
@@ -2997,6 +2998,48 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
final String topicName = "non-persistent://prop-xyz/ns1/testTruncateTopic-" + UUID.randomUUID().toString();
admin.topics().createNonPartitionedTopic(topicName);
assertThrows(() -> {admin.topics().truncate(topicName);});
+ }
+
+ @Test(timeOut = 20000)
+ public void testPeekEncryptedMessages() throws Exception {
+ final String topicName = "persistent://prop-xyz/ns1/testPeekEncryptedMessages-" + UUID.randomUUID().toString();
+ final String subName = "my-sub";
+
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, subName, MessageId.latest);
+
+ final Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topicName)
+ .enableBatching(true)
+ .addEncryptionKey("my-app-key")
+ .defaultCryptoKeyReader("file:./src/test/resources/certificate/public-key.client-rsa.pem")
+ .create();
+
+ for (int i = 0; i < 5; i++) {
+ producer.send(("message-" + i).getBytes());
+ }
+ producer.close();
+
+ final List<Message<byte[]>> peekedMessages = admin.topics().peekMessages(topicName, subName, 5);
+ assertEquals(peekedMessages.size(), 5);
+
+ final Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subName)
+ .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME)
+ .subscribe();
+ final List<Message<byte[]>> receivedMessages = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ Message<byte[]> msg = consumer.receive(5, TimeUnit.SECONDS);
+ receivedMessages.add(msg);
+ consumer.acknowledge(msg);
+ }
+ consumer.unsubscribe();
+
+ for (int i = 0; i < 5; i++) {
+ assertEquals(peekedMessages.get(i).getMessageId(), receivedMessages.get(i).getMessageId());
+ assertEquals(peekedMessages.get(i).getData(), receivedMessages.get(i).getData());
+ }
}
}
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 877448d..647804f 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
@@ -1497,8 +1497,17 @@ public class TopicsImpl extends BaseResource implements Topics {
}
tmp = headers.getFirst(BATCH_HEADER);
- if (response.getHeaderString(BATCH_HEADER) != null) {
+ if (tmp != null) {
properties.put(BATCH_HEADER, (String) tmp);
+ }
+
+ boolean isEncrypted = false;
+ tmp = headers.getFirst("X-Pulsar-Is-Encrypted");
+ if (tmp != null) {
+ isEncrypted = Boolean.parseBoolean(tmp.toString());
+ }
+
+ if (!isEncrypted && response.getHeaderString(BATCH_HEADER) != null) {
return getIndividualMsgsFromBatch(topic, msgId, data, properties, messageMetadata);
}