@Test(expected = IllegalStateException.class) public void cantSubscribeTopicAndPattern() { state.subscribe(singleton(topic), rebalanceListener); state.subscribe(Pattern.compile(".*"), rebalanceListener); }
public void subscribe(Set<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(Set<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(Set<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(Set<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(Set<String> topics, ConsumerRebalanceListener listener); void subscribeFromPattern(Set<String> topics); void groupSubscribe(Collection<String> topics); void resetGroupSubscription(); void assignFromUser(Set<TopicPartition> partitions); void assignFromSubscribed(Collection<TopicPartition> assignments); void subscribe(Pattern pattern, ConsumerRebalanceListener listener); boolean hasPatternSubscription(); boolean hasNoSubscriptionOrUserAssignment(); void unsubscribe(); Pattern subscribedPattern(); 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(); List<TopicPartition> fetchablePartitions(); boolean partitionsAutoAssigned(); void position(TopicPartition tp, long offset); Long position(TopicPartition tp); Long partitionLag(TopicPartition tp, IsolationLevel isolationLevel); void updateHighWatermark(TopicPartition tp, long highWatermark); void updateLastStableOffset(TopicPartition tp, long lastStableOffset); 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(Collection<TopicPartition> partitions); boolean hasAllFetchPositions(); Set<TopicPartition> missingFetchPositions(); boolean isAssigned(TopicPartition tp); boolean isPaused(TopicPartition tp); boolean isFetchable(TopicPartition tp); boolean hasValidPosition(TopicPartition tp); void pause(TopicPartition tp); void resume(TopicPartition tp); void movePartitionToEnd(TopicPartition tp); ConsumerRebalanceListener listener(); void addListener(Listener listener); void fireOnAssignment(Set<TopicPartition> assignment); }
SubscriptionState { public void subscribe(Set<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(Set<String> topics, ConsumerRebalanceListener listener); void subscribeFromPattern(Set<String> topics); void groupSubscribe(Collection<String> topics); void resetGroupSubscription(); void assignFromUser(Set<TopicPartition> partitions); void assignFromSubscribed(Collection<TopicPartition> assignments); void subscribe(Pattern pattern, ConsumerRebalanceListener listener); boolean hasPatternSubscription(); boolean hasNoSubscriptionOrUserAssignment(); void unsubscribe(); Pattern subscribedPattern(); 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(); List<TopicPartition> fetchablePartitions(); boolean partitionsAutoAssigned(); void position(TopicPartition tp, long offset); Long position(TopicPartition tp); Long partitionLag(TopicPartition tp, IsolationLevel isolationLevel); void updateHighWatermark(TopicPartition tp, long highWatermark); void updateLastStableOffset(TopicPartition tp, long lastStableOffset); 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(Collection<TopicPartition> partitions); boolean hasAllFetchPositions(); Set<TopicPartition> missingFetchPositions(); boolean isAssigned(TopicPartition tp); boolean isPaused(TopicPartition tp); boolean isFetchable(TopicPartition tp); boolean hasValidPosition(TopicPartition tp); void pause(TopicPartition tp); void resume(TopicPartition tp); void movePartitionToEnd(TopicPartition tp); ConsumerRebalanceListener listener(); void addListener(Listener listener); void fireOnAssignment(Set<TopicPartition> assignment);  }