if there are many consumers, execution order will be alphabetical list by
Name of @Consumer class.

@param topic
@return
358980
protected Collection loadEvenHandler(String topic) {
		Collection ehs = new ArrayList();
		Collection<String> consumers = (Collection<String>) containerWrapper.lookup(ConsumerLoader.TOPICNAME + topic);
		if (consumers == null || consumers.size() == 0) {
			Debug.logWarning("[Jdonframework]there is no any consumer class annotated with @Consumer(" + topic + ") ", module);
			return ehs;
		}
		for (String consumerName : consumers) {
			DomainEventHandler eh = (DomainEventHandler) containerWrapper.lookup(consumerName);
			ehs.add(eh);
		}

		return ehs;

	}
794254
synchronized List<HandlerHolder> findHandlers(String topic) {
        List<HandlerHolder> handlers = new ArrayList<HandlerHolder>();

        List<HandlerHolder> discreteHandlers = discreteEventHandlers.get(topic);
        if (discreteHandlers != null) {
            handlers.addAll(discreteHandlers);
        }

        for (Map.Entry<String, List<HandlerHolder>> entry : wildcardEventHandlers.entrySet()) {
            if (topic.startsWith(entry.getKey())) {
                handlers.addAll(entry.getValue());
            }
        }

        return handlers;
    }
9183
private void _checkAndSubscribe(KafkaConsumer<?, ?> consumer, String topic) {
        synchronized (consumer) {
            Set<String> subscription = consumer.subscription();
            if (subscription == null || !subscription.contains(topic)) {
                consumer.subscribe(Arrays.asList(topic));
            }
        }
    }
720252
private KafkaMsgConsumerWorker _getWorker(String topic, boolean autoCommitOffsets) {
        KafkaMsgConsumerWorker worker = topicWorkers.get(topic);
        if (worker == null) {
            Collection<IKafkaMessageListener> msgListeners = topicMsgListeners.get(topic);
            worker = new KafkaMsgConsumerWorker(this, topic, msgListeners, executorService);
            KafkaMsgConsumerWorker existingWorker = topicWorkers.putIfAbsent(topic, worker);
            if (existingWorker != null) {
                worker = existingWorker;
            } else {
                worker.start();
            }
        }
        return worker;
    }
1313674
public ServiceFuture<List<TopicInner>> listAsync(final ServiceCallback<List<TopicInner>> serviceCallback) {
        return ServiceFuture.fromResponse(listWithServiceResponseAsync(), serviceCallback);
    }
250632
def get_number_of_messages_in_topics(self, topics):
        

        if not isinstance(topics, list):
            topics = [topics]

        number_of_messages = 0
        for t in topics:
            part = self.get_kafka_partitions_for_topic(topic=t)
            Partitions = map(lambda p: TopicPartition(topic=t, partition=p), part)
            number_of_messages += self.get_number_of_messages_in_topicpartition(Partitions)

        return number_of_messages
5722
protected List<String> getTopicIds() {
        LinkedList<String> topicIds = new LinkedList<String>();
        for (final Entry<String, SpecTopic> specTopicEntry : topics.entrySet()) {
            topicIds.add(specTopicEntry.getKey());
        }
        return topicIds;
    }
363741
private Stream<MemberId> getSubscriberNodes(String topicName) {
    InternalTopic topic = topics.get(topicName);
    if (topic == null) {
      return Stream.empty();
    }
    return topic.remoteSubscriptions().stream()
        .filter(s -> !s.isTombstone())
        .map(s -> s.memberId())
        .distinct();
  }
341892
@Nonnull
  @MustNotContainNull
  public static Topic[] removeSuccessorsAndDuplications(@Nonnull @MustNotContainNull final Topic... topics) {
    final List<Topic> result = new ArrayList<Topic>();

    for (final Topic t : topics) {
      final Iterator<Topic> iterator = result.iterator();
      while (iterator.hasNext()) {
        final Topic listed = iterator.next();
        if (listed == t || listed.hasAncestor(t)) {
          iterator.remove();
        }
      }
      result.add(t);
    }
    return result.toArray(new Topic[result.size()]);
  }
105147
public CMAWebhook addTopic(CMAWebhookTopic topic) {
    if (this.topics == null) {
      this.topics = new ArrayList<CMAWebhookTopic>();
    }

    this.topics.add(topic);
    return this;
  }
295862
public MessageProducer getOrCreateProducer(final String topic) {
        if (!this.shareProducer) {
            FutureTask<MessageProducer> task = this.producers.get(topic);
            if (task == null) {
                task = new FutureTask<MessageProducer>(new Callable<MessageProducer>() {

                    @Override
                    public MessageProducer call() throws Exception {
                        MessageProducer producer = MetaqTemplate.this.messageSessionFactory.createProducer();
                        producer.publish(topic);
                        if (!StringUtils.isBlank(MetaqTemplate.this.defaultTopic)) {
                            producer.setDefaultTopic(MetaqTemplate.this.defaultTopic);
                        }
                        return producer;
                    }

                });
                FutureTask<MessageProducer> oldTask = this.producers.putIfAbsent(topic, task);
                if (oldTask != null) {
                    task = oldTask;
                }
                else {
                    task.run();
                }
            }

            try {
                MessageProducer producer = task.get();
                return producer;
            }
            catch (ExecutionException e) {
                throw ThreadUtils.launderThrowable(e.getCause());
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        else {
            if (this.sharedProducer == null) {
                synchronized (this) {
                    if (this.sharedProducer == null) {
                        this.sharedProducer = this.messageSessionFactory.createProducer();
                        if (!StringUtils.isBlank(this.defaultTopic)) {
                            this.sharedProducer.setDefaultTopic(this.defaultTopic);
                        }
                    }
                }
            }
            this.sharedProducer.publish(topic);
            return this.sharedProducer;
        }
        throw new IllegalStateException("Could not create producer for topic '" + topic + "'");
    }
365218
public static String getTopicName(State state) {
    Preconditions.checkArgument(state.contains(KafkaSource.TOPIC_NAME),
        "Missing configuration property " + KafkaSource.TOPIC_NAME);

    return state.getProp(KafkaSource.TOPIC_NAME);
  }
298210
function getTopicQueue(name) {
  var self = this;
  var queue = self._topicQueues[name];

  if (!queue) {
    // XXX(sam) topic queues should know subscribers and their topics
    queue = self._topicQueues[name] = new EventEmitter();
    queue.publish = function(msg, topic) {
      // XXX(sam) sends to all workers, should only send to ones who subscribe
      // XXX(sam) send same obj to all workers, instead of recreating
      Object.keys(cluster.workers).forEach(function(id) {
        cluster.workers[id].send({
          name: name,
          type: 'publish',
          topic: String(topic),
          msg: msg
        });
      });

      queue.emit('publish', {
        topic: String(topic), // XXX(sam) why String?
        msg: msg
      });
    };
  }

  return queue;
}
923800
public Object[] getResults(String topic)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "getResults", topic);

    if (cachedResults != null)
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "getResults", cachedResults);
      return cachedResults;
    }
    postProcess(topic);

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "getResults", "finalResults: " + Arrays.toString(finalResults));
    return finalResults;
  }
625288
def get_all_topics(region=None, key=None, keyid=None, profile=None):
    '''
    
    '''
    cache_key = _cache_get_key()
    try:
        return __context__[cache_key]
    except KeyError:
        pass

    conn = _get_conn(region=region, key=key, keyid=keyid, profile=profile)
    __context__[cache_key] = {}
    # TODO: support >100 SNS topics (via NextToken)
    topics = conn.get_all_topics()
    for t in topics['ListTopicsResponse']['ListTopicsResult']['Topics']:
        short_name = t['TopicArn'].split(':')[-1]
        __context__[cache_key][short_name] = t['TopicArn']
    return __context__[cache_key]
486320
private MemberId getNextMemberId(String topicName) {
    InternalTopic topic = topics.get(topicName);
    if (topic == null) {
      return null;
    }

    TopicIterator iterator = topic.iterator();
    if (iterator.hasNext()) {
      return iterator.next().memberId();
    }
    return null;
  }
759869
public synchronized List<KafkaDependencyInstanceContext> getDepsByTopicAndEvent(final String topic,
      final String regex) {
    final Map<String, List<KafkaDependencyInstanceContext>> regexMap = this.topicEventMap.get(topic);
    if (regexMap != null) {
      return regexMap.get(regex);
    }
    return Collections.emptyList();
  }