@Test(expected = IllegalStateException.class) public void cantSubscribePatternAndTopic() { state.subscribe(Pattern.compile(".*"), rebalanceListener); state.subscribe(Arrays.asList(topic), rebalanceListener); }
public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) { if (listener == null) { throw new IllegalArgumentException("RebalanceListener cannot be null"); } setSubscriptionType(SubscriptionType.AUTO_TOPICS); this.listener = listener; changeSubscription(topics); }
SubscriptionState { public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) { if (listener == null) { throw new IllegalArgumentException("RebalanceListener cannot be null"); } setSubscriptionType(SubscriptionType.AUTO_TOPICS); this.listener = listener; changeSubscription(topics); } }
SubscriptionState { public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) { if (listener == null) { throw new IllegalArgumentException("RebalanceListener cannot be null"); } setSubscriptionType(SubscriptionType.AUTO_TOPICS); this.listener = listener; changeSubscription(topics); } SubscriptionState(OffsetResetStrategy defaultResetStrategy); }
SubscriptionState { public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) { if (listener == null) { throw new IllegalArgumentException("RebalanceListener cannot be null"); } setSubscriptionType(SubscriptionType.AUTO_TOPICS); this.listener = listener; changeSubscription(topics); } SubscriptionState(OffsetResetStrategy defaultResetStrategy); void subscribe(Collection<String> topics, ConsumerRebalanceListener listener); void changeSubscription(Collection<String> topicsToSubscribe); void groupSubscribe(Collection<String> topics); void needReassignment(); void assignFromUser(Collection<TopicPartition> partitions); void assignFromSubscribed(Collection<TopicPartition> assignments); void subscribe(Pattern pattern, ConsumerRebalanceListener listener); boolean hasPatternSubscription(); void unsubscribe(); Pattern getSubscribedPattern(); Set<String> subscription(); Set<TopicPartition> pausedPartitions(); Set<String> groupSubscription(); void committed(TopicPartition tp, OffsetAndMetadata offset); OffsetAndMetadata committed(TopicPartition tp); void needRefreshCommits(); boolean refreshCommitsNeeded(); void commitsRefreshed(); void seek(TopicPartition tp, long offset); Set<TopicPartition> assignedPartitions(); Set<TopicPartition> fetchablePartitions(); boolean partitionsAutoAssigned(); void position(TopicPartition tp, long offset); Long position(TopicPartition tp); Map<TopicPartition, OffsetAndMetadata> allConsumed(); void needOffsetReset(TopicPartition partition, OffsetResetStrategy offsetResetStrategy); void needOffsetReset(TopicPartition partition); boolean hasDefaultOffsetResetPolicy(); boolean isOffsetResetNeeded(TopicPartition partition); OffsetResetStrategy resetStrategy(TopicPartition partition); boolean hasAllFetchPositions(); Set<TopicPartition> missingFetchPositions(); boolean partitionAssignmentNeeded(); boolean isAssigned(TopicPartition tp); boolean isPaused(TopicPartition tp); boolean isFetchable(TopicPartition tp); void pause(TopicPartition tp); void resume(TopicPartition tp); ConsumerRebalanceListener listener(); }
SubscriptionState { public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) { if (listener == null) { throw new IllegalArgumentException("RebalanceListener cannot be null"); } setSubscriptionType(SubscriptionType.AUTO_TOPICS); this.listener = listener; changeSubscription(topics); } SubscriptionState(OffsetResetStrategy defaultResetStrategy); void subscribe(Collection<String> topics, ConsumerRebalanceListener listener); void changeSubscription(Collection<String> topicsToSubscribe); void groupSubscribe(Collection<String> topics); void needReassignment(); void assignFromUser(Collection<TopicPartition> partitions); void assignFromSubscribed(Collection<TopicPartition> assignments); void subscribe(Pattern pattern, ConsumerRebalanceListener listener); boolean hasPatternSubscription(); void unsubscribe(); Pattern getSubscribedPattern(); Set<String> subscription(); Set<TopicPartition> pausedPartitions(); Set<String> groupSubscription(); void committed(TopicPartition tp, OffsetAndMetadata offset); OffsetAndMetadata committed(TopicPartition tp); void needRefreshCommits(); boolean refreshCommitsNeeded(); void commitsRefreshed(); void seek(TopicPartition tp, long offset); Set<TopicPartition> assignedPartitions(); Set<TopicPartition> fetchablePartitions(); boolean partitionsAutoAssigned(); void position(TopicPartition tp, long offset); Long position(TopicPartition tp); Map<TopicPartition, OffsetAndMetadata> allConsumed(); void needOffsetReset(TopicPartition partition, OffsetResetStrategy offsetResetStrategy); void needOffsetReset(TopicPartition partition); boolean hasDefaultOffsetResetPolicy(); boolean isOffsetResetNeeded(TopicPartition partition); OffsetResetStrategy resetStrategy(TopicPartition partition); boolean hasAllFetchPositions(); Set<TopicPartition> missingFetchPositions(); boolean partitionAssignmentNeeded(); boolean isAssigned(TopicPartition tp); boolean isPaused(TopicPartition tp); boolean isFetchable(TopicPartition tp); void pause(TopicPartition tp); void resume(TopicPartition tp); ConsumerRebalanceListener listener();  }