You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/03/09 04:43:15 UTC
[GitHub] [pulsar] tjiuming opened a new pull request #14616: intercept ACK command by BrokerInterceptor.
tjiuming opened a new pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616
### Motivation
Handle ACK command by `BrokerInterceptor`, so that we can receive ACK command in `BrokerInterceptor#onPulsarCommand`.
### Modifications
Invoke `PulsarDecoder#safeInterceptCommand` before `ServerCnx#handleAck`.
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
- The rest endpoints: (no)
- The admin cli options: (no)
- Anything that affects deployment: (no)
### Documentation
Check the box below or label this PR directly (if you have committer privilege).
Need to update docs?
- [ ] `doc-required`
(If you need help on updating docs, create a doc issue)
- [x] `no-need-doc`
(Please explain why)
- [ ] `doc`
(If this PR contains doc changes)
--
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
[GitHub] [pulsar] tjiuming commented on a change in pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
tjiuming commented on a change in pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#discussion_r826541631
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
##########
@@ -55,6 +55,7 @@
int messageCount = 0;
int messageDispatchCount = 0;
int messageAckCount = 0;
+ int handleAckCount = 0;
Review comment:
The class is for local tests, I think int is ok
--
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
[GitHub] [pulsar] tjiuming commented on a change in pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
tjiuming commented on a change in pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#discussion_r822588673
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerInterceptorTest.java
##########
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
+import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandAck;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.protocol.Commands;
+import org.mockito.Mockito;
+import org.powermock.reflect.Whitebox;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+
+public class BrokerInterceptorTest {
Review comment:
@gaoran10 yes, but I want to mock the `handleAck` method, it's a protected method......
--
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
[GitHub] [pulsar] tjiuming commented on pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
tjiuming commented on pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#issuecomment-1063005678
@codelipenghui @gaoran10 tests improved, PTAL
--
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
[GitHub] [pulsar] codelipenghui merged pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616
--
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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#discussion_r822678852
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerInterceptorTest.java
##########
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
+import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandAck;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.protocol.Commands;
+import org.mockito.Mockito;
+import org.powermock.reflect.Whitebox;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+
+public class BrokerInterceptorTest {
Review comment:
@tjiuming If not able to use the existing `BrokerInterceptorTest`, could you please move the new test class under the `org.apache.pulsar.broker.intercept` package and use a different class name? Otherwise, the new test class most likely not be used in the future.
And the new test class should under the "broker" test group.
BTW, looks like it's easy to improve the existing test `testBeforeSendMessage` to cover the new change, send some messages, consume and ack them later, then check the interceptor works well.
--
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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#discussion_r826513672
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
##########
@@ -179,6 +180,23 @@ public void testBeforeSendMessage() throws PulsarClientException {
assertEquals(((CounterBrokerInterceptor)listener).getMessageDispatchCount(),1);
}
+ @Test
+ public void testInterceptAck() throws Exception {
+ final String topic = "test-intercept-ack" + UUID.randomUUID();
+ BrokerInterceptor interceptor = pulsar.getBrokerInterceptor();
+ Assert.assertTrue(interceptor instanceof CounterBrokerInterceptor);
+
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topic)
+ .subscriptionName("test-sub").subscribe()) {
+ producer.send("test intercept ack message");
+ Message<String> message = consumer.receive();
+ consumer.acknowledge(message);
+ }
+
+ Assert.assertEquals(((CounterBrokerInterceptor) interceptor).getHandleAckCount(), 1);
Review comment:
Need to add `Awaitility.await()` here, for Pulsar consumers, it will not send out the acknowledge command immediately, without some retry policies, here should be a flaky test.
--
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
[GitHub] [pulsar] Technoboy- commented on a change in pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#discussion_r826541129
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
##########
@@ -55,6 +55,7 @@
int messageCount = 0;
int messageDispatchCount = 0;
int messageAckCount = 0;
+ int handleAckCount = 0;
Review comment:
Why are these fields all `int`, not `long` ?
--
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
[GitHub] [pulsar] gaoran10 commented on a change in pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#discussion_r822458250
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerInterceptorTest.java
##########
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
+import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandAck;
+import org.apache.pulsar.common.api.proto.MessageIdData;
+import org.apache.pulsar.common.protocol.Commands;
+import org.mockito.Mockito;
+import org.powermock.reflect.Whitebox;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+
+public class BrokerInterceptorTest {
Review comment:
There is an existing test `BrokerInterceptorTest` in package `org.apache.pulsar.broker.intercept`.
--
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
[GitHub] [pulsar] tjiuming commented on a change in pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
tjiuming commented on a change in pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#discussion_r826533309
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
##########
@@ -179,6 +180,23 @@ public void testBeforeSendMessage() throws PulsarClientException {
assertEquals(((CounterBrokerInterceptor)listener).getMessageDispatchCount(),1);
}
+ @Test
+ public void testInterceptAck() throws Exception {
+ final String topic = "test-intercept-ack" + UUID.randomUUID();
+ BrokerInterceptor interceptor = pulsar.getBrokerInterceptor();
+ Assert.assertTrue(interceptor instanceof CounterBrokerInterceptor);
+
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topic)
+ .subscriptionName("test-sub").subscribe()) {
+ producer.send("test intercept ack message");
+ Message<String> message = consumer.receive();
+ consumer.acknowledge(message);
+ }
+
+ Assert.assertEquals(((CounterBrokerInterceptor) interceptor).getHandleAckCount(), 1);
Review comment:
@codelipenghui I've read org.apache.pulsar.client.api.Consumer before, when consumer is closing, it will invoke AcknowledgmentsGroupingTracker#flush to flush all pending acks to broker. I assert the value handleAckCount after consumer is closed, it seems ok
--
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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #14616: [Feature] intercept ACK command by BrokerInterceptor.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #14616:
URL: https://github.com/apache/pulsar/pull/14616#discussion_r826538548
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
##########
@@ -179,6 +180,23 @@ public void testBeforeSendMessage() throws PulsarClientException {
assertEquals(((CounterBrokerInterceptor)listener).getMessageDispatchCount(),1);
}
+ @Test
+ public void testInterceptAck() throws Exception {
+ final String topic = "test-intercept-ack" + UUID.randomUUID();
+ BrokerInterceptor interceptor = pulsar.getBrokerInterceptor();
+ Assert.assertTrue(interceptor instanceof CounterBrokerInterceptor);
+
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(topic)
+ .subscriptionName("test-sub").subscribe()) {
+ producer.send("test intercept ack message");
+ Message<String> message = consumer.receive();
+ consumer.acknowledge(message);
+ }
+
+ Assert.assertEquals(((CounterBrokerInterceptor) interceptor).getHandleAckCount(), 1);
Review comment:
Ok, I see. Thanks for the explanation.
--
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