[GitHub] [pulsar] eolivelli commented on a diff in pull request #17195: [PIP-180] ShadowTopic - Part II - Add support for messgae_id in CommandSend

[GitBox <gi...@apache.org>]

eolivelli commented on code in PR #17195:
URL: https://github.com/apache/pulsar/pull/17195#discussion_r951776628


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -813,9 +815,17 @@ protected ByteBuf encryptMessage(MessageMetadata msgMetadata, ByteBuf compressed
         }
     }
 
-    protected ByteBufPair sendMessage(long producerId, long sequenceId, int numMessages, MessageMetadata msgMetadata,
-            ByteBuf compressedPayload) {
-        return Commands.newSend(producerId, sequenceId, numMessages, getChecksumType(), msgMetadata, compressedPayload);
+    protected ByteBufPair sendMessage(long producerId, long sequenceId, int numMessages,
+                                      MessageId messageId, MessageMetadata msgMetadata,
+                                      ByteBuf compressedPayload) {
+        if (messageId instanceof MessageIdImpl) {

Review Comment:
   So we have two problems:
   - regular users should not be able to set the MessageId (I don't know if there is a way)
   - the broker must not allow a client to write to a ledger that is not part of the topic



##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -813,9 +815,17 @@ protected ByteBuf encryptMessage(MessageMetadata msgMetadata, ByteBuf compressed
         }
     }
 
-    protected ByteBufPair sendMessage(long producerId, long sequenceId, int numMessages, MessageMetadata msgMetadata,
-            ByteBuf compressedPayload) {
-        return Commands.newSend(producerId, sequenceId, numMessages, getChecksumType(), msgMetadata, compressedPayload);
+    protected ByteBufPair sendMessage(long producerId, long sequenceId, int numMessages,
+                                      MessageId messageId, MessageMetadata msgMetadata,
+                                      ByteBuf compressedPayload) {
+        if (messageId instanceof MessageIdImpl) {

Review Comment:
   This shouldn't be a code path taken by user Pulsar clients but only by the Replicator.
   
   Maybe we should add an internal flag to the PulsarClient to mark it as a internal client and allow it to be able to do this thing.
   Pushing the messageId from the producer also may lead to some security issues.
   We must prevent on the broker side that a client writes to the wrong ledger (for instance to a ledger of another topic, that  can be not accessible from the client according the authz)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org