[GitHub] [pulsar] codelipenghui commented on a diff in pull request #17269: [feature][broker] PIP-204: Extensions for broker interceptor

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

codelipenghui commented on code in PR #17269:
URL: https://github.com/apache/pulsar/pull/17269#discussion_r965982215


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java:
##########
@@ -246,18 +246,26 @@ public boolean checkAndStartPublish(long producerId, long sequenceId, ByteBuf he
 
     private void publishMessageToTopic(ByteBuf headersAndPayload, long sequenceId, long batchSize, boolean isChunked,
                                        boolean isMarker) {
-        topic.publishMessage(headersAndPayload,
-                MessagePublishContext.get(this, sequenceId, msgIn,
-                        headersAndPayload.readableBytes(), batchSize,
-                        isChunked, System.nanoTime(), isMarker));
+        MessagePublishContext messagePublishContext =
+                MessagePublishContext.get(this, sequenceId, msgIn, headersAndPayload.readableBytes(),
+                        batchSize, isChunked, System.nanoTime(), isMarker);
+        if (this.cnx.getBrokerService().getInterceptor() != null){
+            this.cnx.getBrokerService().getInterceptor()
+                    .onMessagePublish(this, headersAndPayload, messagePublishContext);
+        }
+        topic.publishMessage(headersAndPayload, messagePublishContext);
     }
 
     private void publishMessageToTopic(ByteBuf headersAndPayload, long lowestSequenceId, long highestSequenceId,
                                        long batchSize, boolean isChunked, boolean isMarker) {
-        topic.publishMessage(headersAndPayload,
-                MessagePublishContext.get(this, lowestSequenceId,
-                        highestSequenceId, msgIn, headersAndPayload.readableBytes(), batchSize,
-                        isChunked, System.nanoTime(), isMarker));
+        MessagePublishContext messagePublishContext = MessagePublishContext.get(this, lowestSequenceId,
+                highestSequenceId, msgIn, headersAndPayload.readableBytes(), batchSize,
+                isChunked, System.nanoTime(), isMarker);
+        if (this.cnx.getBrokerService().getInterceptor() != null){

Review Comment:
   If we can have a `InterceptorDisabled` implementation, we can avoid many `if` checks like here.



##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java:
##########
@@ -147,6 +147,15 @@ public void beforeSendMessage(Subscription subscription,
                                   Entry entry,
                                   long[] ackSet,
                                   MessageMetadata msgMetadata) {
+        beforeSendMessage(subscription, entry, ackSet, msgMetadata, null);

Review Comment:
   We should not change here.
   I think the current implementation will introduce a breaking change?
   Let's imagine users are using an old interceptor, it will not have this logic.
   
   The expected behavior here is users implemented both 2 method, they will received 2 copies right?



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java:
##########
@@ -133,6 +168,18 @@ public void producerCreated(ServerCnx cnx, Producer producer,
         }
     }
 
+    @Override
+    public void producerClosed(ServerCnx cnx,
+                               Producer producer,
+                               Map<String, String> metadata) {
+        if (interceptors == null || interceptors.isEmpty()) {
+            return;
+        }

Review Comment:
   I see many duplicated code here,
   It's better to have a method that can be reused for all the method
   
   ```java
   if (enabled) {
       for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
               value.producerClosed(cnx, producer, metadata);
           }
   }
   ```



-- 
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