From 9b68d313b563f824059f22daefcef7e2428afb4f Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Tue, 6 Sep 2022 15:15:10 +0200 Subject: [PATCH 1/2] Broker interceptor integration test --- .../plugins/LoggingBrokerInterceptor.java | 128 ++++++++++++++++++ .../META-INF/services/broker_interceptor.yml | 22 +++ .../plugins/TestBrokerInterceptors.java | 114 ++++++++++++++++ .../src/test/resources/pulsar-plugin.xml | 1 + 4 files changed, 265 insertions(+) create mode 100644 tests/docker-images/java-test-plugins/src/main/java/org/apache/pulsar/tests/integration/plugins/LoggingBrokerInterceptor.java create mode 100644 tests/docker-images/java-test-plugins/src/main/resources/META-INF/services/broker_interceptor.yml create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java diff --git a/tests/docker-images/java-test-plugins/src/main/java/org/apache/pulsar/tests/integration/plugins/LoggingBrokerInterceptor.java b/tests/docker-images/java-test-plugins/src/main/java/org/apache/pulsar/tests/integration/plugins/LoggingBrokerInterceptor.java new file mode 100644 index 0000000000000..18c73d21abc59 --- /dev/null +++ b/tests/docker-images/java-test-plugins/src/main/java/org/apache/pulsar/tests/integration/plugins/LoggingBrokerInterceptor.java @@ -0,0 +1,128 @@ +/** + * 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.tests.integration.plugins; + +import io.netty.buffer.ByteBuf; +import java.util.Map; +import javax.servlet.FilterChain; +import javax.servlet.ServletRequest; +import javax.servlet.ServletResponse; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.intercept.BrokerInterceptor; +import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.broker.service.Producer; +import org.apache.pulsar.broker.service.ServerCnx; +import org.apache.pulsar.broker.service.Subscription; +import org.apache.pulsar.broker.service.Topic; +import org.apache.pulsar.common.api.proto.BaseCommand; +import org.apache.pulsar.common.api.proto.CommandAck; +import org.apache.pulsar.common.api.proto.MessageMetadata; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class LoggingBrokerInterceptor implements BrokerInterceptor { + + private final Logger log = LoggerFactory.getLogger(LoggingBrokerInterceptor.class); + + + @Override + public void onPulsarCommand(BaseCommand command, ServerCnx cnx) { + log.info("onPulsarCommand"); + } + + @Override + public void onConnectionClosed(ServerCnx cnx) { + log.info("onConnectionClosed"); + } + + @Override + public void onWebserviceRequest(ServletRequest request) { + log.info("onWebserviceRequest"); + } + + @Override + public void onWebserviceResponse(ServletRequest request, ServletResponse response) { + log.info("onWebserviceResponse"); + } + + @Override + public void initialize(PulsarService pulsarService) { + log.info("initialize: " + (pulsarService != null ? "OK" : "NULL")); + } + + @Override + public void close() { + log.info("close"); + } + + + @Override + public void beforeSendMessage(Subscription subscription, Entry entry, long[] ackSet, MessageMetadata msgMetadata) { + log.info("beforeSendMessage: " + + ("producer".equals(msgMetadata.getProducerName()) ? "OK" : "WRONG")); + } + + @Override + public void onConnectionCreated(ServerCnx cnx) { + log.info("onConnectionCreated"); + } + + @Override + public void producerCreated(ServerCnx cnx, Producer producer, Map metadata) { + log.info("producerCreated"); + } + + @Override + public void consumerCreated(ServerCnx cnx, Consumer consumer, Map metadata) { + log.info("consumerCreated"); + } + + @Override + public void messageProduced(ServerCnx cnx, Producer producer, long startTimeNs, long ledgerId, long entryId, + Topic.PublishContext publishContext) { + log.info("messageProduced"); + } + + @Override + public void messageDispatched(ServerCnx cnx, Consumer consumer, long ledgerId, long entryId, + ByteBuf headersAndPayload) { + log.info("messageDispatched"); + } + + @Override + public void messageAcked(ServerCnx cnx, Consumer consumer, CommandAck ackCmd) { + log.info("messageAcked"); + } + + @Override + public void txnOpened(long tcId, String txnID) { + log.info("txnOpened"); + } + + @Override + public void txnEnded(String txnID, long txnAction) { + log.info("txnEnded"); + } + + @Override + public void onFilter(ServletRequest request, ServletResponse response, FilterChain chain) { + log.info("onFilter"); + } +} diff --git a/tests/docker-images/java-test-plugins/src/main/resources/META-INF/services/broker_interceptor.yml b/tests/docker-images/java-test-plugins/src/main/resources/META-INF/services/broker_interceptor.yml new file mode 100644 index 0000000000000..c90ce3593eab5 --- /dev/null +++ b/tests/docker-images/java-test-plugins/src/main/resources/META-INF/services/broker_interceptor.yml @@ -0,0 +1,22 @@ +# +# 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. +# + +name: loggingInterceptor +description: Broker Interceptor that logs each of its method invocations +interceptorClass: org.apache.pulsar.tests.integration.plugins.LoggingBrokerInterceptor diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java new file mode 100644 index 0000000000000..afb5cbb50a933 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java @@ -0,0 +1,114 @@ +/** + * 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.tests.integration.plugins; + +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; +import lombok.Cleanup; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.ConsumerBuilder; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.tests.integration.messaging.TopicMessagingBase; +import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec; +import org.testng.annotations.Test; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + + +public class TestBrokerInterceptors extends TopicMessagingBase { + + private static final String PREFIX = "PULSAR_PREFIX_"; + + @Override + public void setupCluster() throws Exception { + brokerEnvs.put(PREFIX + "disableBrokerInterceptors", "false"); + brokerEnvs.put(PREFIX + "brokerInterceptorsDirectory", "/pulsar/examples"); + brokerEnvs.put(PREFIX + "brokerInterceptors", "loggingInterceptor"); + super.setupCluster(); + } + + @Override + protected PulsarClusterSpec.PulsarClusterSpecBuilder beforeSetupCluster(String clusterName, + PulsarClusterSpec.PulsarClusterSpecBuilder specBuilder) { + specBuilder.numBrokers(1); + return specBuilder; + } + + @Test(dataProvider = "ServiceUrls") + public void test(Supplier serviceUrlSupplier) throws Exception { + String serviceUrl = serviceUrlSupplier.get(); + + final String topicName = getNonPartitionedTopic("interceptorTest-topic", true); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName("producer") + .create(); + int messagesToSend = 20; + for (int i = 0; i < messagesToSend; i++) { + String messageValue = producer.getProducerName() + "-" + i; + MessageId messageId = producer.newMessage() + .value(messageValue) + .send(); + assertNotNull(messageId); + } + + try (Consumer consumer = createConsumer(client, topicName)) { + for (int i = 0; i < messagesToSend; ++i) { + consumer.receive(3, TimeUnit.SECONDS); + } + } + + String log = pulsarCluster.getAnyBroker() + .execCmd("cat", "/var/log/pulsar/broker.log").getStdout(); + + for (String line : new String[]{ + "initialize: OK", + "onConnectionCreated", + "producerCreated", + "consumerCreated", + "messageProduced", + "beforeSendMessage: OK", + }) { + assertTrue(log.contains("LoggingBrokerInterceptor - " + line), "Log did not contain line '" + line + "'"); + } + + } + + private Consumer createConsumer(PulsarClient client, String topicName) throws Exception { + ConsumerBuilder builder = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName(randomName(8)) + .subscriptionType(SubscriptionType.Exclusive) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + return builder.subscribe(); + } +} diff --git a/tests/integration/src/test/resources/pulsar-plugin.xml b/tests/integration/src/test/resources/pulsar-plugin.xml index a93cac164dded..957d63f7fe82e 100644 --- a/tests/integration/src/test/resources/pulsar-plugin.xml +++ b/tests/integration/src/test/resources/pulsar-plugin.xml @@ -23,6 +23,7 @@ + From 4a8056a6030ca845ea5a6e35d28b6cd8f6ee0287 Mon Sep 17 00:00:00 2001 From: Andras Beni Date: Wed, 7 Sep 2022 15:05:21 +0200 Subject: [PATCH 2/2] Remove unused import --- .../pulsar/tests/integration/plugins/TestBrokerInterceptors.java | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java index afb5cbb50a933..3f5f02e5b9374 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java @@ -23,7 +23,6 @@ import lombok.Cleanup; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.ConsumerBuilder; -import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient;