eolivelli commented on a change in pull request #12970: URL: https://github.com/apache/pulsar/pull/12970#discussion_r759015193
########## File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/plugin/EntryFilterTest.java ########## @@ -0,0 +1,48 @@ +/** + * 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.plugin; + + +import java.util.List; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.pulsar.common.api.proto.KeyValue; + +public class EntryFilterTest implements EntryFilter { + @Override + public FilterResult filterEntry(Entry entry, FilterContext context) { + if (context.getMsgMetadata() == null || context.getMsgMetadata().getPropertiesCount() <= 0) { + return FilterResult.ACCEPT; + } + List<KeyValue> list = context.getMsgMetadata().getPropertiesList(); + // filter by string + for (KeyValue keyValue : list) { + if ("ACCEPT".equalsIgnoreCase(keyValue.getKey())) { + return FilterResult.ACCEPT; + } else if ("REJECT".equalsIgnoreCase(keyValue.getKey())){ + return FilterResult.REJECT; + } + } + return null; Review comment: what happens in case of "null" result ? ########## File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/plugin/EntryFilterProvider.java ########## @@ -0,0 +1,152 @@ +/** + * 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.plugin; + +import static com.google.common.base.Preconditions.checkArgument; +import com.google.common.collect.ImmutableMap; +import java.io.File; +import java.io.IOException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Collections; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.common.nar.NarClassLoader; +import org.apache.pulsar.common.util.ObjectMapperFactory; + +@Slf4j +public class EntryFilterProvider { + + static final String ENTRY_FILTER_DEFINITION_FILE = "entry_filter.yml"; + + /** + * create entry filter instance. + */ + public static ImmutableMap<String, EntryFilterWithClassLoader> createEntryFilters( + ServiceConfiguration conf) throws IOException { + EntryFilterDefinitions definitions = searchForEntryFilters(conf.getEntryFiltersDirectory(), + conf.getNarExtractionDirectory()); + ImmutableMap.Builder<String, EntryFilterWithClassLoader> builder = ImmutableMap.builder(); + for (String filterName : conf.getEntryFilterNames()) { + EntryFilterMetaData metaData = definitions.getFilters().get(filterName); + if (null == metaData) { + throw new RuntimeException("No entry filter is found for name `" + filterName + + "`. Available entry filters are : " + definitions.getFilters()); + } + EntryFilterWithClassLoader filter; + filter = load(metaData, conf.getNarExtractionDirectory()); + if (filter != null) { + builder.put(filterName, filter); + } + log.info("Successfully loaded entry filter for name `{}`", filterName); + } + return builder.build(); + } + + private static EntryFilterDefinitions searchForEntryFilters(String entryFiltersDirectory, + String narExtractionDirectory) + throws IOException { + Path path = Paths.get(entryFiltersDirectory).toAbsolutePath(); + log.info("Searching for entry filters in {}", path); + + EntryFilterDefinitions entryFilterDefinitions = new EntryFilterDefinitions(); + if (!path.toFile().exists()) { + log.warn("Pulsar entry filters directory not found"); Review comment: log.info ########## File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/plugin/EntryFilterTest.java ########## @@ -0,0 +1,48 @@ +/** + * 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.plugin; + + +import java.util.List; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.pulsar.common.api.proto.KeyValue; + +public class EntryFilterTest implements EntryFilter { + @Override + public FilterResult filterEntry(Entry entry, FilterContext context) { + if (context.getMsgMetadata() == null || context.getMsgMetadata().getPropertiesCount() <= 0) { + return FilterResult.ACCEPT; + } + List<KeyValue> list = context.getMsgMetadata().getPropertiesList(); + // filter by string + for (KeyValue keyValue : list) { + if ("ACCEPT".equalsIgnoreCase(keyValue.getKey())) { + return FilterResult.ACCEPT; + } else if ("REJECT".equalsIgnoreCase(keyValue.getKey())){ + return FilterResult.REJECT; + } + } + return null; + } + + @Override + public void close() { + Review comment: can we verify that the close() method has been called ? you can use a static variable ########## File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java ########## @@ -183,12 +215,35 @@ public int filterEntriesForConsumer(Optional<EntryWrapper[]> entryWrapper, int e interceptor.beforeSendMessage(subscription, entry, ackSet, msgMetadata); } } + if (CollectionUtils.isNotEmpty(entriesToFiltered)) { + subscription.acknowledgeMessage(entriesToFiltered, AckType.Individual, + Collections.emptyMap()); + } + sendMessageInfo.setTotalMessages(totalMessages); sendMessageInfo.setTotalBytes(totalBytes); sendMessageInfo.setTotalChunkedMessages(totalChunkedMessages); return totalEntries; } + private EntryFilter.FilterResult getFilterResult(FilterContext filterContext, Entry entry) { Review comment: nit: static -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
