public ConsumerMessageFilter findFilter(final String topic, final String group) {
        if (this.filterClassLoader == null) {
            return null;
        }
        final String className = this.metaConfig.getTopicConfig(topic).getFilterClass(group);
        if (StringUtils.isBlank(className)) {
            return null;
        }
        FutureTask<ConsumerMessageFilter> task = this.filters.get(className);
        if (task == null) {
            task = new FutureTask<ConsumerMessageFilter>(new Callable<ConsumerMessageFilter>() {

                @Override
                public ConsumerMessageFilter call() throws Exception {
                    ConsumerMessageFilter intanceFilter = ConsumerFilterManager.this.intanceFilter(className);
                    if (intanceFilter != null) {
                        log.warn("Created filter '" + className + "' for group:" + group + " and topic:" + topic);
                    }
                    return intanceFilter;
                }

            });
            FutureTask<ConsumerMessageFilter> existsTask = this.filters.putIfAbsent(className, task);
            if (existsTask != null) {
                task = existsTask;
            }
            else {
                task.run();
            }
        }
        return this.getFilter0(task);
    }
Consumer filter manager.

@author dennis<killme2008@gmail.com>
metamorphosis-server/src/main/java/com/taobao/metamorphosis/server/filter/ConsumerFilterManager.java