[improve][pip] PIP-432: Add isEncrypted field to EncryptionContext (apache#24481)

Author: codelipenghui

pip/pip-432 Add isEncrypted field to EncryptionContext.md

@@ -0,0 +1,120 @@
+# PIP-432: Add isEncrypted field to EncryptionContext
+
+# Background knowledge
+
+Apache Pulsar supports client-side encryption where messages can be encrypted by producers and decrypted by consumers. When a message is encrypted, Pulsar includes an `EncryptionContext` with each message that contains encryption metadata such as:
+
+- **Encryption keys**: The encrypted data encryption keys used for message encryption
+- **Encryption parameters**: Additional parameters like initialization vectors (IV)
+- **Encryption algorithm**: The algorithm used (e.g., RSA, ECDSA)
+- **Compression information**: Whether compression was applied before encryption
+
+**Key concepts:**
+- **EncryptionContext**: A metadata object attached to encrypted messages containing encryption-related information
+- **CryptoKeyReader**: An interface that provides public/private keys for encryption/decryption operations
+- **ConsumerCryptoFailureAction**: Determines how consumers handle decryption failures:
+  - `FAIL`: Fail message consumption (default)
+  - `DISCARD`: Silently discard the message
+  - `CONSUME`: Deliver the encrypted message to the application
+
+Currently, when `ConsumerCryptoFailureAction.CONSUME` is configured, consumers can receive encrypted messages even when decryption fails (e.g., missing private key, mismatched keys). However, applications have no way to determine whether the received message was successfully decrypted or is still encrypted.
+
+# Motivation
+
+Applications using Pulsar's encryption feature with `ConsumerCryptoFailureAction.CONSUME` need to determine whether received messages were successfully decrypted or if decryption failed. This is essential for:
+
+1. **Error handling**: Applications need to know when they receive encrypted (undecrypted) data to handle it appropriately
+2. **Monitoring**: Applications want to track decryption success/failure rates for monitoring and alerting
+3. **Manual decryption**: When automatic decryption fails, applications may want to attempt manual decryption using the EncryptionContext
+4. **Security compliance**: Applications need to ensure they're not inadvertently processing encrypted data as plain text
+
+**Current situation:**
+- Consumers with `CONSUME` action receive messages regardless of decryption success
+- No programmatic way to distinguish between successfully decrypted and failed decryption messages
+- Applications must implement workarounds to detect encrypted vs. decrypted content
+
+**Use cases this solves:**
+1. Consumer without private key configured → should know decryption failed
+2. Consumer with mismatched private key → should know decryption failed  
+3. Consumer with correct private key → should know decryption succeeded
+
+# Goals
+
+## In Scope
+
+- Add an `isEncrypted` boolean field to the `EncryptionContext` class
+- Update consumer decryption logic to populate this field correctly
+- Ensure the field accurately reflects decryption status for all encryption scenarios
+- Maintain backward compatibility with existing applications
+- Update existing encryption tests to verify the new functionality
+
+## Out of Scope
+
+- Changes to encryption/decryption algorithms or protocols
+- Modifications to `ConsumerCryptoFailureAction` behavior
+- Performance improvements to encryption/decryption operations
+- New encryption features or capabilities
+- Changes to producer-side encryption logic
+
+# High Level Design
+
+The solution adds a simple boolean field `isEncrypted` to the existing `EncryptionContext` class. This field is set during message processing in the consumer:
+
+1. **Successful decryption**: When a consumer successfully decrypts a message, `isEncrypted` is set to `false`
+2. **Failed decryption**: When decryption fails (missing key, wrong key, etc.) but `ConsumerCryptoFailureAction.CONSUME` is configured, `isEncrypted` is set to `true`
+3. **No encryption**: For non-encrypted messages, no `EncryptionContext` is created (existing behavior)
+
+The field is populated in the consumer's message creation logic, specifically in the `createEncryptionContext()` method where the decryption success/failure status is already known.
+
+Applications can then check this field to determine if the received message payload is encrypted or decrypted:
+
+```java
+Message<byte[]> message = consumer.receive();
+Optional<EncryptionContext> ctx = message.getEncryptionCtx();
+if (ctx.isPresent()) {
+    if (ctx.get().isEncrypted()) {
+        // Handle encrypted message - decryption failed
+        handleEncryptedMessage(message, ctx.get());
+    } else {
+        // Handle decrypted message - decryption succeeded
+        handleDecryptedMessage(message);
+    }
+}
+```
+
+## Public-facing Changes
+
+### Public API
+
+**New method available via Lombok-generated getter:**
+```java
+public boolean isEncrypted()
+```
+
+**Usage pattern:**
+```java
+Message<T> message = consumer.receive();
+Optional<EncryptionContext> encryptionCtx = message.getEncryptionCtx();
+if (encryptionCtx.isPresent()) {
+    boolean encrypted = encryptionCtx.get().isEncrypted();
+    if (encrypted) {
+        // Message is encrypted (decryption failed)
+    } else {
+        // Message is decrypted (decryption succeeded)
+    }
+}
+```
+
+**Breaking Changes:** None - this is a purely additive change.
+
+# Backward & Forward Compatibility
+
+No compatibility concerns.
+
+## Downgrade / Rollback
+
+Rolling back to a previous Pulsar version:
+1. Applications using `isEncrypted()` will get compilation errors - this is expected
+2. Remove calls to `isEncrypted()` from application code
+3. Downgrade Pulsar client library
+4. No data loss or corruption risk

pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java

@@ -3297,6 +3297,11 @@ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMe
             String expectedMessage = "my-message-" + msgNum++;
             Assert.assertNotEquals(receivedMessage, expectedMessage, "Received encrypted message " + receivedMessage
                     + " should not match the expected message " + expectedMessage);
+            // Verify that decryption failed since no CryptoKeyReader is configured
+            Optional<EncryptionContext> encryptionCtx = msg.getEncryptionCtx();
+            Assert.assertTrue(encryptionCtx.isPresent(), "EncryptionContext should be present for encrypted message");
+            Assert.assertTrue(encryptionCtx.get().isEncrypted(),
+                "isEncrypted should be true when no CryptoKeyReader is configured");
             consumer.acknowledgeCumulative(msg);
         } catch (Exception e) {
             e.printStackTrace();
@@ -3313,8 +3318,12 @@ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMe
         for (int i = msgNum; i < totalMsg - 1; i++) {
             msg = (MessageImpl<byte[]>) consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
             // verify that encrypted message contains encryption-context
-            msg.getEncryptionCtx()
+            Optional<EncryptionContext> encryptionCtx = msg.getEncryptionCtx();
+            EncryptionContext ctx = encryptionCtx
                     .orElseThrow(() -> new IllegalStateException("encryption-ctx not present for encrypted message"));
+            // Verify that decryption succeeded since CryptoKeyReader is properly configured
+            Assert.assertFalse(ctx.isEncrypted(),
+                "isEncrypted should be false when CryptoKeyReader is properly configured and decryption succeeds");
             String receivedMessage = new String(msg.getData());
             log.debug("Received message: [{}]", receivedMessage);
             String expectedMessage = "my-message-" + i;
@@ -3399,6 +3408,12 @@ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMe
         TopicMessageImpl<byte[]> msg =
                 (TopicMessageImpl<byte[]>) consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
 
+        // Verify that decryption failed since no CryptoKeyReader is configured for the consumer
+        Optional<EncryptionContext> encryptionCtx = msg.getEncryptionCtx();
+        Assert.assertTrue(encryptionCtx.isPresent(), "EncryptionContext should be present for encrypted message");
+        Assert.assertTrue(encryptionCtx.get().isEncrypted(),
+            "isEncrypted should be true when consumer has no CryptoKeyReader configured");
+
         String receivedMessage = decryptMessage(msg, encryptionKeyName, new EncKeyReader());
         assertEquals(message, receivedMessage);
 
@@ -5250,6 +5265,15 @@ public void testE2EEncryptionWithCompression() throws Exception {
         for (int i = 0; i < 10; i++) {
             final var msg = consumer.receive(5, TimeUnit.SECONDS);
             assertNotNull(msg);
+            // Verify that decryption failed due to mismatched keys
+            if (msg instanceof MessageImpl) {
+                MessageImpl<String> msgImpl = (MessageImpl<String>) msg;
+                Optional<EncryptionContext> encryptionCtx = msgImpl.getEncryptionCtx();
+                Assert.assertTrue(encryptionCtx.isPresent(),
+                        "EncryptionContext should be present for encrypted message");
+                Assert.assertTrue(encryptionCtx.get().isEncrypted(),
+                    "isEncrypted should be true when using mismatched keys for decryption");
+            }
             consumer.acknowledge(msg);
         }
 

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/CompactedOutBatchMessageTest.java

@@ -82,7 +82,7 @@ public void testCompactedOutMessages() throws Exception {
             // shove it in the sideways
             consumer.receiveIndividualMessagesFromBatch(brokerEntryMetadata, metadata, 0, null,
                     batchBuffer, new MessageIdData().setLedgerId(1234).setEntryId(567),
-                    consumer.cnx(), DEFAULT_CONSUMER_EPOCH);
+                    consumer.cnx(), DEFAULT_CONSUMER_EPOCH, false);
             Message<?> m = consumer.receive();
             assertEquals(((BatchMessageIdImpl) m.getMessageId()).getLedgerId(), 1234);
             assertEquals(((BatchMessageIdImpl) m.getMessageId()).getEntryId(), 567);

pulsar-client-api/src/main/java/org/apache/pulsar/common/api/EncryptionContext.java

@@ -39,6 +39,8 @@ public class EncryptionContext {
     private CompressionType compressionType;
     private int uncompressedMessageSize;
     private Optional<Integer> batchSize;
+    // Indicates whether the message payload remains encrypted (true) or has been successfully decrypted (false)
+    private boolean isEncrypted;
 
     /**
      * Encryption key with metadata.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -1290,7 +1290,8 @@ protected <V> MessageImpl<V> newSingleMessage(final int index,
                                                   final BitSetRecyclable ackBitSet,
                                                   final BitSet ackSetInMessageId,
                                                   final int redeliveryCount,
-                                                  final long consumerEpoch) {
+                                                  final long consumerEpoch,
+                                                  final boolean isEncrypted) {
         if (log.isDebugEnabled()) {
             log.debug("[{}] [{}] processing message num - {} in batch", subscription, consumerName, index);
         }
@@ -1328,7 +1329,8 @@ protected <V> MessageImpl<V> newSingleMessage(final int index,
             final ByteBuf payloadBuffer = (singleMessagePayload != null) ? singleMessagePayload : payload;
             final MessageImpl<V> message = MessageImpl.create(topicName.toString(), batchMessageIdImpl,
                     msgMetadata, singleMessageMetadata, payloadBuffer,
-                    createEncryptionContext(msgMetadata), cnx(), schema, redeliveryCount, poolMessages, consumerEpoch);
+                    createEncryptionContext(msgMetadata, isEncrypted), cnx(), schema, redeliveryCount,
+                    poolMessages, consumerEpoch);
             message.setBrokerEntryMetadata(brokerEntryMetadata);
             return message;
         } catch (IOException | IllegalStateException e) {
@@ -1347,8 +1349,21 @@ protected <V> MessageImpl<V> newMessage(final MessageIdImpl messageId,
                                             final Schema<V> schema,
                                             final int redeliveryCount,
                                             final long consumerEpoch) {
+        return newMessage(messageId, brokerEntryMetadata, messageMetadata, payload, schema, redeliveryCount,
+                consumerEpoch, false);
+    }
+
+    protected <V> MessageImpl<V> newMessage(final MessageIdImpl messageId,
+                                            final BrokerEntryMetadata brokerEntryMetadata,
+                                            final MessageMetadata messageMetadata,
+                                            final ByteBuf payload,
+                                            final Schema<V> schema,
+                                            final int redeliveryCount,
+                                            final long consumerEpoch,
+                                            final boolean isEncrypted) {
         final MessageImpl<V> message = MessageImpl.create(topicName.toString(), messageId, messageMetadata, payload,
-                createEncryptionContext(messageMetadata), cnx(), schema, redeliveryCount, poolMessages, consumerEpoch);
+                createEncryptionContext(messageMetadata, isEncrypted), cnx(), schema, redeliveryCount,
+                poolMessages, consumerEpoch);
         message.setBrokerEntryMetadata(brokerEntryMetadata);
         return message;
     }
@@ -1532,7 +1547,7 @@ void messageReceived(CommandMessage cmdMessage, ByteBuf headersAndPayload, Clien
 
             final MessageImpl<T> message =
                     newMessage(msgId, brokerEntryMetadata, msgMetadata, uncompressedPayload,
-                            schema, redeliveryCount, consumerEpoch);
+                            schema, redeliveryCount, consumerEpoch, isMessageUndecryptable);
             uncompressedPayload.release();
 
             if (deadLetterPolicy != null && possibleSendToDeadLetterTopicMessages != null) {
@@ -1552,7 +1567,7 @@ void messageReceived(CommandMessage cmdMessage, ByteBuf headersAndPayload, Clien
         } else {
             // handle batch message enqueuing; uncompressed payload has all messages in batch
             receiveIndividualMessagesFromBatch(brokerEntryMetadata, msgMetadata, redeliveryCount, ackSet,
-                    uncompressedPayload, messageId, cnx, consumerEpoch);
+                    uncompressedPayload, messageId, cnx, consumerEpoch, isMessageUndecryptable);
 
             uncompressedPayload.release();
         }
@@ -1752,7 +1767,8 @@ private void interceptAndComplete(final Message<T> message, final CompletableFut
 
     void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata, MessageMetadata msgMetadata,
                                             int redeliveryCount, List<Long> ackSet, ByteBuf uncompressedPayload,
-                                            MessageIdData messageId, ClientCnx cnx, long consumerEpoch) {
+                                            MessageIdData messageId, ClientCnx cnx, long consumerEpoch,
+                                            boolean isEncrypted) {
         int batchSize = msgMetadata.getNumMessagesInBatch();
 
         // create ack tracker for entry aka batch
@@ -1775,7 +1791,7 @@ void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata,
             for (int i = 0; i < batchSize; ++i) {
                 final MessageImpl<T> message = newSingleMessage(i, batchSize, brokerEntryMetadata, msgMetadata,
                         singleMessageMetadata, uncompressedPayload, batchMessage, schema, true,
-                        ackBitSet, ackSetInMessageId, redeliveryCount, consumerEpoch);
+                        ackBitSet, ackSetInMessageId, redeliveryCount, consumerEpoch, isEncrypted);
                 if (message == null) {
                     // If it is not in ackBitSet, it means Broker does not want to deliver it to the client, and
                     // did not decrease the permits in the broker-side.
@@ -2905,9 +2921,11 @@ private boolean isMessageUndecryptable(MessageMetadata msgMetadata) {
      * Create EncryptionContext if message payload is encrypted.
      *
      * @param msgMetadata
+     * @param isEncrypted
      * @return {@link Optional}<{@link EncryptionContext}>
      */
-    private Optional<EncryptionContext> createEncryptionContext(MessageMetadata msgMetadata) {
+    private Optional<EncryptionContext> createEncryptionContext(MessageMetadata msgMetadata,
+                                                                boolean isEncrypted) {
 
         EncryptionContext encryptionCtx = null;
         if (msgMetadata.getEncryptionKeysCount() > 0) {
@@ -2930,6 +2948,7 @@ private Optional<EncryptionContext> createEncryptionContext(MessageMetadata msgM
                     .setCompressionType(CompressionCodecProvider.convertFromWireProtocol(msgMetadata.getCompression()));
             encryptionCtx.setUncompressedMessageSize(msgMetadata.getUncompressedSize());
             encryptionCtx.setBatchSize(batchSize);
+            encryptionCtx.setEncrypted(isEncrypted);
         }
         return Optional.ofNullable(encryptionCtx);
     }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessagePayloadContextImpl.java

@@ -137,7 +137,8 @@ public <T> Message<T> getMessageAt(int index,
                     ackBitSet,
                     ackSetInMessageId,
                     redeliveryCount,
-                    consumerEpoch);
+                    consumerEpoch,
+                    false);
         } finally {
             payloadBuffer.release();
         }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java

@@ -194,7 +194,8 @@ protected void tryTriggerListener() {
     @Override
     void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata, MessageMetadata msgMetadata,
                                             int redeliveryCount, List<Long> ackSet, ByteBuf uncompressedPayload,
-                                            MessageIdData messageId, ClientCnx cnx, long consumerEpoch) {
+                                            MessageIdData messageId, ClientCnx cnx, long consumerEpoch,
+                                            boolean isEncrypted) {
         log.warn(
                 "Closing consumer [{}]-[{}] due to unsupported received batch-message with zero receiver queue size",
                 subscription, consumerName);