[pulsar] branch branch-2.9 updated: [fix][flaky-test] BrokerInterceptorTest.testProducerCreation (#16742)

[bogong]

This is an automated email from the ASF dual-hosted git repository.

bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.9 by this push:
     new 54561f0ad70 [fix][flaky-test] 
BrokerInterceptorTest.testProducerCreation (#16742)
54561f0ad70 is described below

commit 54561f0ad70d0284d15e5c6b1283f1e043bf2c82
Author: Baodi Shi <wudixiaolong...@icloud.com>
AuthorDate: Mon Jul 25 14:05:27 2022 +0800

    [fix][flaky-test] BrokerInterceptorTest.testProducerCreation (#16742)
    
    (cherry picked from commit bc94643bc1a7f365dfb75e389ac3e1156770a119)
---
 .../broker/intercept/CounterBrokerInterceptor.java | 45 ++++++++++++++--------
 1 file changed, 30 insertions(+), 15 deletions(-)

diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
index 1462cfab895..fe5c080ceab 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.intercept;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 import javax.servlet.FilterChain;
 import javax.servlet.ServletException;
 import javax.servlet.ServletRequest;
@@ -40,9 +41,15 @@ import org.eclipse.jetty.server.Response;
 @Slf4j
 public class CounterBrokerInterceptor implements BrokerInterceptor {
 
-    int beforeSendCount = 0;
-    int count = 0;
-    private List<ResponseEvent> responseList = new ArrayList<>();
+    private final AtomicInteger beforeSendCount = new AtomicInteger();
+    private final AtomicInteger count = new AtomicInteger();
+
+    public void reset() {
+        beforeSendCount.set(0);
+        count.set(0);
+    }
+
+    private final List<ResponseEvent> responseList = new ArrayList<>();
 
     @Data
     @AllArgsConstructor
@@ -56,15 +63,19 @@ public class CounterBrokerInterceptor implements 
BrokerInterceptor {
                                   Entry entry,
                                   long[] ackSet,
                                   MessageMetadata msgMetadata) {
-        log.debug("Send message to topic {}, subscription {}",
-            subscription.getTopic(), subscription.getName());
-        beforeSendCount++;
+        if (log.isDebugEnabled()) {
+            log.debug("Send message to topic {}, subscription {}",
+                    subscription.getTopic(), subscription.getName());
+        }
+        beforeSendCount.incrementAndGet();
     }
 
     @Override
     public void onPulsarCommand(BaseCommand command, ServerCnx cnx) {
-        log.debug("[{}] On [{}] Pulsar command", count, 
command.getType().name());
-        count ++;
+        if (log.isDebugEnabled()) {
+            log.debug("[{}] On [{}] Pulsar command", count, 
command.getType().name());
+        }
+        count.incrementAndGet();
     }
 
     @Override
@@ -74,14 +85,18 @@ public class CounterBrokerInterceptor implements 
BrokerInterceptor {
 
     @Override
     public void onWebserviceRequest(ServletRequest request) {
-        count ++;
-        log.debug("[{}] On [{}] Webservice request", count, 
((HttpServletRequest)request).getRequestURL().toString());
+        count.incrementAndGet();
+        if (log.isDebugEnabled()) {
+            log.debug("[{}] On [{}] Webservice request", count, 
((HttpServletRequest) request).getRequestURL().toString());
+        }
     }
 
     @Override
     public void onWebserviceResponse(ServletRequest request, ServletResponse 
response) {
-        count ++;
-        log.debug("[{}] On [{}] Webservice response {}", count, 
((HttpServletRequest)request).getRequestURL().toString(), response);
+        count.incrementAndGet();
+        if (log.isDebugEnabled()) {
+            log.debug("[{}] On [{}] Webservice response {}", count, 
((HttpServletRequest) request).getRequestURL().toString(), response);
+        }
         if (response instanceof Response) {
             Response res = (Response) response;
             responseList.add(new 
ResponseEvent(res.getHttpChannel().getRequest().getRequestURI(), 
res.getStatus()));
@@ -91,7 +106,7 @@ public class CounterBrokerInterceptor implements 
BrokerInterceptor {
     @Override
     public void onFilter(ServletRequest request, ServletResponse response, 
FilterChain chain)
             throws IOException, ServletException {
-        count = 100;
+        count.set(100);
         chain.doFilter(request, response);
     }
 
@@ -106,11 +121,11 @@ public class CounterBrokerInterceptor implements 
BrokerInterceptor {
     }
 
     public int getCount() {
-        return count;
+        return count.get();
     }
 
     public int getBeforeSendCount() {
-        return beforeSendCount;
+        return beforeSendCount.get();
     }
 
     public void clearResponseList() {