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..3f5f02e5b9374 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestBrokerInterceptors.java @@ -0,0 +1,113 @@ +/** + * 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.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 c03b3e8007d35..f88b67256e547 100644 --- a/tests/integration/src/test/resources/pulsar-plugin.xml +++ b/tests/integration/src/test/resources/pulsar-plugin.xml @@ -24,6 +24,7 @@ +