@Test public void testInterceptorConstructorClose() throws Exception { try { Properties props = new Properties(); props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); props.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MockConsumerInterceptor.class.getName()); KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>( props, new StringDeserializer(), new StringDeserializer()); assertEquals(1, MockConsumerInterceptor.INIT_COUNT.get()); assertEquals(0, MockConsumerInterceptor.CLOSE_COUNT.get()); consumer.close(); assertEquals(1, MockConsumerInterceptor.INIT_COUNT.get()); assertEquals(1, MockConsumerInterceptor.CLOSE_COUNT.get()); Assert.assertNull(MockConsumerInterceptor.CLUSTER_META.get()); } finally { MockConsumerInterceptor.resetCounters(); } }
@Override public void close() { close(DEFAULT_CLOSE_TIMEOUT_MS, TimeUnit.MILLISECONDS); }
KafkaConsumer implements Consumer<K, V> { @Override public void close() { close(DEFAULT_CLOSE_TIMEOUT_MS, TimeUnit.MILLISECONDS); } }
KafkaConsumer implements Consumer<K, V> { @Override public void close() { close(DEFAULT_CLOSE_TIMEOUT_MS, TimeUnit.MILLISECONDS); } KafkaConsumer(Map<String, Object> configs);  KafkaConsumer(Map<String, Object> configs,
                         Deserializer<K> keyDeserializer,
                         Deserializer<V> valueDeserializer);  KafkaConsumer(Properties properties);  KafkaConsumer(Properties properties,
                         Deserializer<K> keyDeserializer,
                         Deserializer<V> valueDeserializer); @SuppressWarnings("unchecked") private  KafkaConsumer(ConsumerConfig config,
                          Deserializer<K> keyDeserializer,
                          Deserializer<V> valueDeserializer);   KafkaConsumer(String clientId,
                  ConsumerCoordinator coordinator,
                  Deserializer<K> keyDeserializer,
                  Deserializer<V> valueDeserializer,
                  Fetcher<K, V> fetcher,
                  ConsumerInterceptors<K, V> interceptors,
                  Time time,
                  ConsumerNetworkClient client,
                  Metrics metrics,
                  SubscriptionState subscriptions,
                  Metadata metadata,
                  long retryBackoffMs,
                  long requestTimeoutMs); }
KafkaConsumer implements Consumer<K, V> { @Override public void close() { close(DEFAULT_CLOSE_TIMEOUT_MS, TimeUnit.MILLISECONDS); } KafkaConsumer(Map<String, Object> configs);  KafkaConsumer(Map<String, Object> configs,
                         Deserializer<K> keyDeserializer,
                         Deserializer<V> valueDeserializer);  KafkaConsumer(Properties properties);  KafkaConsumer(Properties properties,
                         Deserializer<K> keyDeserializer,
                         Deserializer<V> valueDeserializer); @SuppressWarnings("unchecked") private  KafkaConsumer(ConsumerConfig config,
                          Deserializer<K> keyDeserializer,
                          Deserializer<V> valueDeserializer);   KafkaConsumer(String clientId,
                  ConsumerCoordinator coordinator,
                  Deserializer<K> keyDeserializer,
                  Deserializer<V> valueDeserializer,
                  Fetcher<K, V> fetcher,
                  ConsumerInterceptors<K, V> interceptors,
                  Time time,
                  ConsumerNetworkClient client,
                  Metrics metrics,
                  SubscriptionState subscriptions,
                  Metadata metadata,
                  long retryBackoffMs,
                  long requestTimeoutMs); Set<TopicPartition> assignment(); Set<String> subscription(); @Override // 订阅指定的topic,并未消费者自动分配分区 void subscribe(Collection<String> topics, ConsumerRebalanceListener listener); @Override void subscribe(Collection<String> topics); @Override void subscribe(Pattern pattern, ConsumerRebalanceListener listener); void unsubscribe(); @Override // 用户手动订阅指定的topic并指定消费的分区，和subscribe方法互斥 void assign(Collection<TopicPartition> partitions); @Override // 负责从服务端获取消息 ConsumerRecords<K, V> poll(long timeout); @Override void commitSync(); @Override void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets); @Override void commitAsync(); @Override void commitAsync(OffsetCommitCallback callback); @Override void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback); @Override // 指定消费者起始消费的位置 void seek(TopicPartition partition, long offset); void seekToBeginning(Collection<TopicPartition> partitions); void seekToEnd(Collection<TopicPartition> partitions); long position(TopicPartition partition); @Override OffsetAndMetadata committed(TopicPartition partition); @Override Map<MetricName, ? extends Metric> metrics(); @Override List<PartitionInfo> partitionsFor(String topic); @Override Map<String, List<PartitionInfo>> listTopics(); @Override void pause(Collection<TopicPartition> partitions); @Override // 恢复consumer void resume(Collection<TopicPartition> partitions); @Override // 暂停consumer Set<TopicPartition> paused(); @Override Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch); @Override Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions); @Override Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions); @Override void close(); void close(long timeout, TimeUnit timeUnit); @Override void wakeup(); }
KafkaConsumer implements Consumer<K, V> { @Override public void close() { close(DEFAULT_CLOSE_TIMEOUT_MS, TimeUnit.MILLISECONDS); } KafkaConsumer(Map<String, Object> configs);  KafkaConsumer(Map<String, Object> configs,
                         Deserializer<K> keyDeserializer,
                         Deserializer<V> valueDeserializer);  KafkaConsumer(Properties properties);  KafkaConsumer(Properties properties,
                         Deserializer<K> keyDeserializer,
                         Deserializer<V> valueDeserializer); @SuppressWarnings("unchecked") private  KafkaConsumer(ConsumerConfig config,
                          Deserializer<K> keyDeserializer,
                          Deserializer<V> valueDeserializer);   KafkaConsumer(String clientId,
                  ConsumerCoordinator coordinator,
                  Deserializer<K> keyDeserializer,
                  Deserializer<V> valueDeserializer,
                  Fetcher<K, V> fetcher,
                  ConsumerInterceptors<K, V> interceptors,
                  Time time,
                  ConsumerNetworkClient client,
                  Metrics metrics,
                  SubscriptionState subscriptions,
                  Metadata metadata,
                  long retryBackoffMs,
                  long requestTimeoutMs); Set<TopicPartition> assignment(); Set<String> subscription(); @Override // 订阅指定的topic,并未消费者自动分配分区 void subscribe(Collection<String> topics, ConsumerRebalanceListener listener); @Override void subscribe(Collection<String> topics); @Override void subscribe(Pattern pattern, ConsumerRebalanceListener listener); void unsubscribe(); @Override // 用户手动订阅指定的topic并指定消费的分区，和subscribe方法互斥 void assign(Collection<TopicPartition> partitions); @Override // 负责从服务端获取消息 ConsumerRecords<K, V> poll(long timeout); @Override void commitSync(); @Override void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets); @Override void commitAsync(); @Override void commitAsync(OffsetCommitCallback callback); @Override void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback); @Override // 指定消费者起始消费的位置 void seek(TopicPartition partition, long offset); void seekToBeginning(Collection<TopicPartition> partitions); void seekToEnd(Collection<TopicPartition> partitions); long position(TopicPartition partition); @Override OffsetAndMetadata committed(TopicPartition partition); @Override Map<MetricName, ? extends Metric> metrics(); @Override List<PartitionInfo> partitionsFor(String topic); @Override Map<String, List<PartitionInfo>> listTopics(); @Override void pause(Collection<TopicPartition> partitions); @Override // 恢复consumer void resume(Collection<TopicPartition> partitions); @Override // 暂停consumer Set<TopicPartition> paused(); @Override Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch); @Override Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions); @Override Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions); @Override void close(); void close(long timeout, TimeUnit timeUnit); @Override void wakeup();  }