| /** |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE |
| * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file |
| * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the |
| * License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
| * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations under the License. |
| */ |
| package org.apache.kafka.clients.consumer; |
| |
| import org.apache.kafka.common.Metric; |
| import org.apache.kafka.common.TopicPartition; |
| import org.apache.kafka.common.metrics.JmxReporter; |
| import org.apache.kafka.common.metrics.MetricConfig; |
| import org.apache.kafka.common.MetricName; |
| import org.apache.kafka.common.metrics.Metrics; |
| import org.apache.kafka.common.metrics.MetricsReporter; |
| import org.apache.kafka.common.serialization.Deserializer; |
| import org.apache.kafka.common.utils.ClientUtils; |
| import org.apache.kafka.common.utils.SystemTime; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.net.InetSocketAddress; |
| import java.util.*; |
| |
| /** |
| * A Kafka client that consumes records from a Kafka cluster. |
| * <P> |
| * The consumer is <i>thread safe</i> and should generally be shared among all threads for best performance. |
| * <p> |
| * The consumer is single threaded and multiplexes I/O over TCP connections to each of the brokers it |
| * needs to communicate with. Failure to close the consumer after use will leak these resources. |
| * <h3>Usage Examples</h3> |
| * The consumer APIs offer flexibility to cover a variety of consumption use cases. Following are some examples to demonstrate the correct use of |
| * the available APIs. Each of the examples assumes the presence of a user implemented process() method that processes a given batch of messages |
| * and returns the offset of the latest processed message per partition. Note that process() is not part of the consumer API and is only used as |
| * a convenience method to demonstrate the different use cases of the consumer APIs. Here is a sample implementation of such a process() method. |
| * <pre> |
| * {@code |
| * private Map<TopicPartition, Long> process(Map<String, ConsumerRecord<byte[], byte[]> records) { |
| * Map<TopicPartition, Long> processedOffsets = new HashMap<TopicPartition, Long>(); |
| * for(Entry<String, ConsumerRecords<byte[], byte[]>> recordMetadata : records.entrySet()) { |
| * List<ConsumerRecord<byte[], byte[]>> recordsPerTopic = recordMetadata.getValue().records(); |
| * for(int i = 0;i < recordsPerTopic.size();i++) { |
| * ConsumerRecord<byte[], byte[]> record = recordsPerTopic.get(i); |
| * // process record |
| * try { |
| * processedOffsets.put(record.topicAndpartition(), record.offset()); |
| * } catch (Exception e) { |
| * e.printStackTrace(); |
| * } |
| * } |
| * } |
| * return processedOffsets; |
| * } |
| * } |
| * </pre> |
| * <p> |
| * This example demonstrates how the consumer can be used to leverage Kafka's group management functionality for automatic consumer load |
| * balancing and failover. This example assumes that the offsets are stored in Kafka and are automatically committed periodically, |
| * as controlled by the auto.commit.interval.ms config |
| * <pre> |
| * {@code |
| * Properties props = new Properties(); |
| * props.put("metadata.broker.list", "localhost:9092"); |
| * props.put("group.id", "test"); |
| * props.put("session.timeout.ms", "1000"); |
| * props.put("enable.auto.commit", "true"); |
| * props.put("auto.commit.interval.ms", "10000"); |
| * KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props); |
| * consumer.subscribe("foo", "bar"); |
| * boolean isRunning = true; |
| * while(isRunning) { |
| * Map<String, ConsumerRecords<byte[], byte[]>> records = consumer.poll(100); |
| * process(records); |
| * } |
| * consumer.close(); |
| * } |
| * </pre> |
| * This example demonstrates how the consumer can be used to leverage Kafka's group management functionality for automatic consumer load |
| * balancing and failover. This example assumes that the offsets are stored in Kafka and are manually committed using |
| * the commit(boolean) API. This example also demonstrates rewinding the consumer's offsets if processing of the consumed |
| * messages fails. Note that this method of rewinding offsets using {@link #seek(Map) seek(offsets)} is only useful for rewinding the offsets |
| * of the current consumer instance. As such, this will not trigger a rebalance or affect the fetch offsets for the other consumer instances. |
| * <pre> |
| * {@code |
| * Properties props = new Properties(); |
| * props.put("metadata.broker.list", "localhost:9092"); |
| * props.put("group.id", "test"); |
| * props.put("session.timeout.ms", "1000"); |
| * props.put("enable.auto.commit", "false"); |
| * KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props); |
| * consumer.subscribe("foo", "bar"); |
| * int commitInterval = 100; |
| * int numRecords = 0; |
| * boolean isRunning = true; |
| * Map<TopicPartition, Long> consumedOffsets = new HashMap<TopicPartition, Long>(); |
| * while(isRunning) { |
| * Map<String, ConsumerRecords<byte[], byte[]>> records = consumer.poll(100); |
| * try { |
| * Map<TopicPartition, Long> lastConsumedOffsets = process(records); |
| * consumedOffsets.putAll(lastConsumedOffsets); |
| * numRecords += records.size(); |
| * // commit offsets for all partitions of topics foo, bar synchronously, owned by this consumer instance |
| * if(numRecords % commitInterval == 0) |
| * consumer.commit(false); |
| * } catch(Exception e) { |
| * try { |
| * // rewind consumer's offsets for failed partitions |
| * // assume failedPartitions() returns the list of partitions for which the processing of the last batch of messages failed |
| * List<TopicPartition> failedPartitions = failedPartitions(); |
| * Map<TopicPartition, Long> offsetsToRewindTo = new HashMap<TopicPartition, Long>(); |
| * for(TopicPartition failedPartition : failedPartitions) { |
| * // rewind to the last consumed offset for the failed partition. Since process() failed for this partition, the consumed offset |
| * // should still be pointing to the last successfully processed offset and hence is the right offset to rewind consumption to. |
| * offsetsToRewindTo.put(failedPartition, consumedOffsets.get(failedPartition)); |
| * } |
| * // seek to new offsets only for partitions that failed the last process() |
| * consumer.seek(offsetsToRewindTo); |
| * } catch(Exception e) { break; } // rewind failed |
| * } |
| * } |
| * consumer.close(); |
| * } |
| * </pre> |
| * <p> |
| * This example demonstrates how to rewind the offsets of the entire consumer group. It is assumed that the user has chosen to use Kafka's |
| * group management functionality for automatic consumer load balancing and failover. This example also assumes that the offsets are stored in |
| * Kafka. If group management is used, the right place to systematically rewind offsets for <i>every</i> consumer instance is inside the |
| * ConsumerRebalanceCallback. The onPartitionsAssigned callback is invoked after the consumer is assigned a new set of partitions on rebalance |
| * <i>and</i> before the consumption restarts post rebalance. This is the right place to supply the newly rewound offsets to the consumer. It |
| * is recommended that if you foresee the requirement to ever reset the consumer's offsets in the presence of group management, that you |
| * always configure the consumer to use the ConsumerRebalanceCallback with a flag that protects whether or not the offset rewind logic is used. |
| * This method of rewinding offsets is useful if you notice an issue with your message processing after successful consumption and offset commit. |
| * And you would like to rewind the offsets for the entire consumer group as part of rolling out a fix to your processing logic. In this case, |
| * you would configure each of your consumer instances with the offset rewind configuration flag turned on and bounce each consumer instance |
| * in a rolling restart fashion. Each restart will trigger a rebalance and eventually all consumer instances would have rewound the offsets for |
| * the partitions they own, effectively rewinding the offsets for the entire consumer group. |
| * <pre> |
| * {@code |
| * Properties props = new Properties(); |
| * props.put("metadata.broker.list", "localhost:9092"); |
| * props.put("group.id", "test"); |
| * props.put("session.timeout.ms", "1000"); |
| * props.put("enable.auto.commit", "false"); |
| * KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>( |
| * props, |
| * new ConsumerRebalanceCallback() { |
| * boolean rewindOffsets = true; // should be retrieved from external application config |
| * public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) { |
| * Map<TopicPartition, Long> latestCommittedOffsets = consumer.committed(partitions); |
| * if(rewindOffsets) |
| * Map<TopicPartition, Long> newOffsets = rewindOffsets(latestCommittedOffsets, 100); |
| * consumer.seek(newOffsets); |
| * } |
| * public void onPartitionsRevoked(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) { |
| * consumer.commit(true); |
| * } |
| * // this API rewinds every partition back by numberOfMessagesToRewindBackTo messages |
| * private Map<TopicPartition, Long> rewindOffsets(Map<TopicPartition, Long> currentOffsets, |
| * long numberOfMessagesToRewindBackTo) { |
| * Map<TopicPartition, Long> newOffsets = new HashMap<TopicPartition, Long>(); |
| * for(Map.Entry<TopicPartition, Long> offset : currentOffsets.entrySet()) |
| * newOffsets.put(offset.getKey(), offset.getValue() - numberOfMessagesToRewindBackTo); |
| * return newOffsets; |
| * } |
| * }); |
| * consumer.subscribe("foo", "bar"); |
| * int commitInterval = 100; |
| * int numRecords = 0; |
| * boolean isRunning = true; |
| * Map<TopicPartition, Long> consumedOffsets = new HashMap<TopicPartition, Long>(); |
| * while(isRunning) { |
| * Map<String, ConsumerRecords<byte[], byte[]>> records = consumer.poll(100); |
| * Map<TopicPartition, Long> lastConsumedOffsets = process(records); |
| * consumedOffsets.putAll(lastConsumedOffsets); |
| * numRecords += records.size(); |
| * // commit offsets for all partitions of topics foo, bar synchronously, owned by this consumer instance |
| * if(numRecords % commitInterval == 0) |
| * consumer.commit(consumedOffsets, true); |
| * } |
| * consumer.commit(true); |
| * consumer.close(); |
| * } |
| * </pre> |
| * This example demonstrates how the consumer can be used to leverage Kafka's group management functionality along with custom offset storage. |
| * In this example, the assumption made is that the user chooses to store the consumer offsets outside Kafka. This requires the user to |
| * plugin logic for retrieving the offsets from a custom store and provide the offsets to the consumer in the ConsumerRebalanceCallback |
| * callback. The onPartitionsAssigned callback is invoked after the consumer is assigned a new set of partitions on rebalance <i>and</i> |
| * before the consumption restarts post rebalance. This is the right place to supply offsets from a custom store to the consumer. |
| * <p> |
| * Similarly, the user would also be required to plugin logic for storing the consumer's offsets to a custom store. The onPartitionsRevoked |
| * callback is invoked right after the consumer has stopped fetching data and before the partition ownership changes. This is the right place |
| * to commit the offsets for the current set of partitions owned by the consumer. |
| * <pre> |
| * {@code |
| * Properties props = new Properties(); |
| * props.put("metadata.broker.list", "localhost:9092"); |
| * props.put("group.id", "test"); |
| * props.put("session.timeout.ms", "1000"); |
| * props.put("enable.auto.commit", "false"); // since enable.auto.commit only applies to Kafka based offset storage |
| * KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>( |
| * props, |
| * new ConsumerRebalanceCallback() { |
| * public void onPartitionsAssigned(Consumer<?,?> consumer, Collection<TopicPartition> partitions) { |
| * Map<TopicPartition, Long> lastCommittedOffsets = getLastCommittedOffsetsFromCustomStore(partitions); |
| * consumer.seek(lastCommittedOffsets); |
| * } |
| * public void onPartitionsRevoked(Consumer<?,?> consumer, Collection<TopicPartition> partitions) { |
| * Map<TopicPartition, Long> offsets = getLastConsumedOffsets(partitions); |
| * commitOffsetsToCustomStore(offsets); |
| * } |
| * // following APIs should be implemented by the user for custom offset management |
| * private Map<TopicPartition, Long> getLastCommittedOffsetsFromCustomStore(Collection<TopicPartition> partitions) { |
| * return null; |
| * } |
| * private Map<TopicPartition, Long> getLastConsumedOffsets(Collection<TopicPartition> partitions) { return null; } |
| * private void commitOffsetsToCustomStore(Map<TopicPartition, Long> offsets) {} |
| * }); |
| * consumer.subscribe("foo", "bar"); |
| * int commitInterval = 100; |
| * int numRecords = 0; |
| * boolean isRunning = true; |
| * Map<TopicPartition, Long> consumedOffsets = new HashMap<TopicPartition, Long>(); |
| * while(isRunning) { |
| * Map<String, ConsumerRecords<byte[], byte[]>> records = consumer.poll(100); |
| * Map<TopicPartition, Long> lastConsumedOffsets = process(records); |
| * consumedOffsets.putAll(lastConsumedOffsets); |
| * numRecords += records.size(); |
| * // commit offsets for all partitions of topics foo, bar synchronously, owned by this consumer instance |
| * if(numRecords % commitInterval == 0) |
| * commitOffsetsToCustomStore(consumedOffsets); |
| * } |
| * consumer.commit(true); |
| * consumer.close(); |
| * } |
| * </pre> |
| * This example demonstrates how the consumer can be used to subscribe to specific partitions of certain topics and consume upto the latest |
| * available message for each of those partitions before shutting down. When used to subscribe to specific partitions, the user foregoes |
| * the group management functionality and instead relies on manually configuring the consumer instances to subscribe to a set of partitions. |
| * This example assumes that the user chooses to use Kafka based offset storage. The user still has to specify a group.id to use Kafka |
| * based offset management. However, session.timeout.ms is not required since the Kafka consumer only does automatic failover when group |
| * management is used. |
| * <pre> |
| * {@code |
| * Properties props = new Properties(); |
| * props.put("metadata.broker.list", "localhost:9092"); |
| * props.put("group.id", "test"); |
| * props.put("enable.auto.commit", "true"); |
| * props.put("auto.commit.interval.ms", "10000"); |
| * KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props); |
| * // subscribe to some partitions of topic foo |
| * TopicPartition partition0 = new TopicPartition("foo", 0); |
| * TopicPartition partition1 = new TopicPartition("foo", 1); |
| * TopicPartition[] partitions = new TopicPartition[2]; |
| * partitions[0] = partition0; |
| * partitions[1] = partition1; |
| * consumer.subscribe(partitions); |
| * // find the last committed offsets for partitions 0,1 of topic foo |
| * Map<TopicPartition, Long> lastCommittedOffsets = consumer.committed(Arrays.asList(partitions)); |
| * // seek to the last committed offsets to avoid duplicates |
| * consumer.seek(lastCommittedOffsets); |
| * // find the offsets of the latest available messages to know where to stop consumption |
| * Map<TopicPartition, Long> latestAvailableOffsets = consumer.offsetsBeforeTime(-2, Arrays.asList(partitions)); |
| * boolean isRunning = true; |
| * Map<TopicPartition, Long> consumedOffsets = new HashMap<TopicPartition, Long>(); |
| * while(isRunning) { |
| * Map<String, ConsumerRecords<byte[], byte[]>> records = consumer.poll(100); |
| * Map<TopicPartition, Long> lastConsumedOffsets = process(records); |
| * consumedOffsets.putAll(lastConsumedOffsets); |
| * for(TopicPartition partition : partitions) { |
| * if(consumedOffsets.get(partition) >= latestAvailableOffsets.get(partition)) |
| * isRunning = false; |
| * else |
| * isRunning = true; |
| * } |
| * } |
| * consumer.commit(true); |
| * consumer.close(); |
| * } |
| * </pre> |
| * This example demonstrates how the consumer can be used to subscribe to specific partitions of certain topics and consume upto the latest |
| * available message for each of those partitions before shutting down. When used to subscribe to specific partitions, the user foregoes |
| * the group management functionality and instead relies on manually configuring the consumer instances to subscribe to a set of partitions. |
| * This example assumes that the user chooses to use custom offset storage. |
| * <pre> |
| * {@code |
| * Properties props = new Properties(); |
| * props.put("metadata.broker.list", "localhost:9092"); |
| * KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props); |
| * // subscribe to some partitions of topic foo |
| * TopicPartition partition0 = new TopicPartition("foo", 0); |
| * TopicPartition partition1 = new TopicPartition("foo", 1); |
| * TopicPartition[] partitions = new TopicPartition[2]; |
| * partitions[0] = partition0; |
| * partitions[1] = partition1; |
| * consumer.subscribe(partitions); |
| * Map<TopicPartition, Long> lastCommittedOffsets = getLastCommittedOffsetsFromCustomStore(); |
| * // seek to the last committed offsets to avoid duplicates |
| * consumer.seek(lastCommittedOffsets); |
| * // find the offsets of the latest available messages to know where to stop consumption |
| * Map<TopicPartition, Long> latestAvailableOffsets = consumer.offsetsBeforeTime(-2, Arrays.asList(partitions)); |
| * boolean isRunning = true; |
| * Map<TopicPartition, Long> consumedOffsets = new HashMap<TopicPartition, Long>(); |
| * while(isRunning) { |
| * Map<String, ConsumerRecords<byte[], byte[]>> records = consumer.poll(100); |
| * Map<TopicPartition, Long> lastConsumedOffsets = process(records); |
| * consumedOffsets.putAll(lastConsumedOffsets); |
| * // commit offsets for partitions 0,1 for topic foo to custom store |
| * commitOffsetsToCustomStore(consumedOffsets); |
| * for(TopicPartition partition : partitions) { |
| * if(consumedOffsets.get(partition) >= latestAvailableOffsets.get(partition)) |
| * isRunning = false; |
| * else |
| * isRunning = true; |
| * } |
| * } |
| * commitOffsetsToCustomStore(consumedOffsets); |
| * consumer.close(); |
| * } |
| * </pre> |
| */ |
| public class KafkaConsumer<K,V> implements Consumer<K,V> { |
| |
| private static final Logger log = LoggerFactory.getLogger(KafkaConsumer.class); |
| |
| private final long metadataFetchTimeoutMs; |
| private final long totalMemorySize; |
| private final Metrics metrics; |
| private final Set<String> subscribedTopics; |
| private final Set<TopicPartition> subscribedPartitions; |
| private final Deserializer<K> keyDeserializer; |
| private final Deserializer<V> valueDeserializer; |
| |
| /** |
| * A consumer is instantiated by providing a set of key-value pairs as configuration. Valid configuration strings |
| * are documented <a href="http://kafka.apache.org/documentation.html#consumerconfigs">here</a>. Values can be |
| * either strings or Objects of the appropriate type (for example a numeric configuration would accept either the |
| * string "42" or the integer 42). |
| * <p> |
| * Valid configuration strings are documented at {@link ConsumerConfig} |
| * @param configs The consumer configs |
| */ |
| public KafkaConsumer(Map<String, Object> configs) { |
| this(configs, null); |
| } |
| |
| /** |
| * A consumer is instantiated by providing a set of key-value pairs as configuration and a {@link ConsumerRebalanceCallback} |
| * implementation |
| * <p> |
| * Valid configuration strings are documented at {@link ConsumerConfig} |
| * @param configs The consumer configs |
| * @param callback A callback interface that the user can implement to manage customized offsets on the start and end of |
| * every rebalance operation. |
| */ |
| public KafkaConsumer(Map<String, Object> configs, ConsumerRebalanceCallback callback) { |
| this(configs, callback, null, null); |
| } |
| |
| /** |
| * A consumer is instantiated by providing a set of key-value pairs as configuration, a {@link ConsumerRebalanceCallback} |
| * implementation, a key and a value {@link Deserializer}. |
| * <p> |
| * Valid configuration strings are documented at {@link ConsumerConfig} |
| * @param configs The consumer configs |
| * @param callback A callback interface that the user can implement to manage customized offsets on the start and end of |
| * every rebalance operation. |
| * @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method won't |
| * be called when the deserializer is passed in directly. |
| * @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method |
| * won't be called when the deserializer is passed in directly. |
| */ |
| public KafkaConsumer(Map<String, Object> configs, ConsumerRebalanceCallback callback, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) { |
| this(new ConsumerConfig(addDeserializerToConfig(configs, keyDeserializer, valueDeserializer)), |
| callback, keyDeserializer, valueDeserializer); |
| } |
| |
| private static Map<String, Object> addDeserializerToConfig(Map<String, Object> configs, |
| Deserializer<?> keyDeserializer, Deserializer<?> valueDeserializer) { |
| Map<String, Object> newConfigs = new HashMap<String, Object>(); |
| newConfigs.putAll(configs); |
| if (keyDeserializer != null) |
| newConfigs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass()); |
| if (keyDeserializer != null) |
| newConfigs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass()); |
| return newConfigs; |
| } |
| |
| /** |
| * A consumer is instantiated by providing a {@link java.util.Properties} object as configuration. |
| * Valid configuration strings are documented at {@link ConsumerConfig} |
| */ |
| public KafkaConsumer(Properties properties) { |
| this(properties, null); |
| } |
| |
| /** |
| * A consumer is instantiated by providing a {@link java.util.Properties} object as configuration and a |
| * {@link ConsumerRebalanceCallback} implementation. |
| * <p> |
| * Valid configuration strings are documented at {@link ConsumerConfig} |
| * @param properties The consumer configuration properties |
| * @param callback A callback interface that the user can implement to manage customized offsets on the start and end of |
| * every rebalance operation. |
| */ |
| public KafkaConsumer(Properties properties, ConsumerRebalanceCallback callback) { |
| this(properties, callback, null, null); |
| } |
| |
| /** |
| * A consumer is instantiated by providing a {@link java.util.Properties} object as configuration and a |
| * {@link ConsumerRebalanceCallback} implementation, a key and a value {@link Deserializer}. |
| * <p> |
| * Valid configuration strings are documented at {@link ConsumerConfig} |
| * @param properties The consumer configuration properties |
| * @param callback A callback interface that the user can implement to manage customized offsets on the start and end of |
| * every rebalance operation. |
| * @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method won't |
| * be called when the deserializer is passed in directly. |
| * @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method |
| * won't be called when the deserializer is passed in directly. |
| */ |
| public KafkaConsumer(Properties properties, ConsumerRebalanceCallback callback, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) { |
| this(new ConsumerConfig(addDeserializerToConfig(properties, keyDeserializer, valueDeserializer)), |
| callback, keyDeserializer, valueDeserializer); |
| } |
| |
| private static Properties addDeserializerToConfig(Properties properties, |
| Deserializer<?> keyDeserializer, Deserializer<?> valueDeserializer) { |
| Properties newProperties = new Properties(); |
| newProperties.putAll(properties); |
| if (keyDeserializer != null) |
| newProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName()); |
| if (keyDeserializer != null) |
| newProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName()); |
| return newProperties; |
| } |
| |
| private KafkaConsumer(ConsumerConfig config, ConsumerRebalanceCallback callback, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) { |
| log.trace("Starting the Kafka consumer"); |
| subscribedTopics = new HashSet<String>(); |
| subscribedPartitions = new HashSet<TopicPartition>(); |
| this.metrics = new Metrics(new MetricConfig(), |
| Collections.singletonList((MetricsReporter) new JmxReporter("kafka.consumer.")), |
| new SystemTime()); |
| this.metadataFetchTimeoutMs = config.getLong(ConsumerConfig.METADATA_FETCH_TIMEOUT_CONFIG); |
| this.totalMemorySize = config.getLong(ConsumerConfig.TOTAL_BUFFER_MEMORY_CONFIG); |
| List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)); |
| |
| if (keyDeserializer == null) |
| this.keyDeserializer = config.getConfiguredInstance(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, |
| Deserializer.class); |
| else |
| this.keyDeserializer = keyDeserializer; |
| if (valueDeserializer == null) |
| this.valueDeserializer = config.getConfiguredInstance(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, |
| Deserializer.class); |
| else |
| this.valueDeserializer = valueDeserializer; |
| |
| config.logUnused(); |
| log.debug("Kafka consumer started"); |
| } |
| |
| /** |
| * Incrementally subscribes to the given list of topics and uses the consumer's group management functionality |
| * <p> |
| * As part of group management, the consumer will keep track of the list of consumers that belong to a particular group and |
| * will trigger a rebalance operation if one of the following events trigger - |
| * <ul> |
| * <li> Number of partitions change for any of the subscribed list of topics |
| * <li> Topic is created or deleted |
| * <li> An existing member of the consumer group dies |
| * <li> A new member is added to an existing consumer group via the join API |
| * </ul> |
| * @param topics A variable list of topics that the consumer wants to subscribe to |
| */ |
| @Override |
| public void subscribe(String... topics) { |
| if(subscribedPartitions.size() > 0) |
| throw new IllegalStateException("Subcription to topics and partitions is mutually exclusive"); |
| for(String topic:topics) |
| subscribedTopics.add(topic); |
| // TODO: trigger a rebalance operation |
| } |
| |
| /** |
| * Incrementally subscribes to a specific topic partition and does not use the consumer's group management functionality. As such, |
| * there will be no rebalance operation triggered when group membership or cluster and topic metadata change. |
| * <p> |
| * @param partitions Partitions to incrementally subscribe to |
| */ |
| @Override |
| public void subscribe(TopicPartition... partitions) { |
| if(subscribedTopics.size() > 0) |
| throw new IllegalStateException("Subcription to topics and partitions is mutually exclusive"); |
| for(TopicPartition partition:partitions) |
| subscribedPartitions.add(partition); |
| } |
| |
| /** |
| * Unsubscribe from the specific topics. This will trigger a rebalance operation and messages for this topic will not be returned |
| * from the next {@link #poll(long) poll()} onwards |
| * @param topics Topics to unsubscribe from |
| */ |
| public void unsubscribe(String... topics) { |
| // throw an exception if the topic was never subscribed to |
| for(String topic:topics) { |
| if(!subscribedTopics.contains(topic)) |
| throw new IllegalStateException("Topic " + topic + " was never subscribed to. subscribe(" + topic + ") should be called prior" + |
| " to unsubscribe(" + topic + ")"); |
| subscribedTopics.remove(topic); |
| } |
| // TODO trigger a rebalance operation |
| } |
| |
| /** |
| * Unsubscribe from the specific topic partitions. Messages for these partitions will not be returned from the next |
| * {@link #poll(long) poll()} onwards |
| * @param partitions Partitions to unsubscribe from |
| */ |
| public void unsubscribe(TopicPartition... partitions) { |
| // throw an exception if the partition was never subscribed to |
| for(TopicPartition partition:partitions) { |
| if(!subscribedPartitions.contains(partition)) |
| throw new IllegalStateException("Partition " + partition + " was never subscribed to. subscribe(new TopicPartition(" + |
| partition.topic() + "," + partition.partition() + ") should be called prior" + |
| " to unsubscribe(new TopicPartition(" + partition.topic() + "," + partition.partition() + ")"); |
| subscribedPartitions.remove(partition); |
| } |
| // trigger a rebalance operation |
| } |
| |
| /** |
| * Fetches data for the topics or partitions specified using one of the subscribe APIs. It is an error to not have subscribed to |
| * any topics or partitions before polling for data. |
| * <p> |
| * The offset used for fetching the data is governed by whether or not {@link #seek(Map) seek(offsets)} |
| * is used. If {@link #seek(Map) seek(offsets)} is used, it will use the specified offsets on startup and |
| * on every rebalance, to consume data from that offset sequentially on every poll. If not, it will use the last checkpointed offset |
| * using {@link #commit(Map, boolean) commit(offsets, sync)} |
| * for the subscribed list of partitions. |
| * @param timeout The time, in milliseconds, spent waiting in poll if data is not available. If 0, waits indefinitely. Must not be negative |
| * @return map of topic to records since the last fetch for the subscribed list of topics and partitions |
| */ |
| @Override |
| public Map<String, ConsumerRecords<K,V>> poll(long timeout) { |
| // TODO Auto-generated method stub |
| return null; |
| } |
| |
| /** |
| * Commits the specified offsets for the specified list of topics and partitions to Kafka. |
| * <p> |
| * This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after every rebalance |
| * and also on startup. As such, if you need to store offsets in anything other than Kafka, this API should not be used. |
| * @param offsets The list of offsets per partition that should be committed to Kafka. |
| * @param sync If true, commit will block until the consumer receives an acknowledgment |
| * @return An {@link OffsetMetadata} object that contains the partition, offset and a corresponding error code. Returns null |
| * if the sync flag is set to false. |
| */ |
| @Override |
| public OffsetMetadata commit(Map<TopicPartition, Long> offsets, boolean sync) { |
| throw new UnsupportedOperationException(); |
| } |
| |
| /** |
| * Commits offsets returned on the last {@link #poll(long) poll()} for the subscribed list of topics and |
| * partitions. |
| * <p> |
| * This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after every rebalance |
| * and also on startup. As such, if you need to store offsets in anything other than Kafka, this API should not be used. |
| * @param sync If true, commit will block until the consumer receives an acknowledgment |
| * @return An {@link OffsetMetadata} object that contains the partition, offset and a corresponding error code. Returns null |
| * if the sync flag is set to false. |
| */ |
| @Override |
| public OffsetMetadata commit(boolean sync) { |
| throw new UnsupportedOperationException(); |
| } |
| |
| /** |
| * Overrides the fetch offsets that the consumer will use on the next {@link #poll(long) poll(timeout)}. If this API is invoked |
| * for the same partition more than once, the latest offset will be used on the next poll(). Note that you may lose data if this API is |
| * arbitrarily used in the middle of consumption, to reset the fetch offsets |
| */ |
| @Override |
| public void seek(Map<TopicPartition, Long> offsets) { |
| } |
| |
| /** |
| * Returns the fetch position of the <i>next message</i> for the specified topic partition to be used on the next {@link #poll(long) poll()} |
| * @param partitions Partitions for which the fetch position will be returned |
| * @return The position from which data will be fetched for the specified partition on the next {@link #poll(long) poll()} |
| */ |
| public Map<TopicPartition, Long> position(Collection<TopicPartition> partitions) { |
| return null; |
| } |
| |
| /** |
| * Fetches the last committed offsets of partitions that the consumer currently consumes. This API is only relevant if Kafka based offset |
| * storage is used. This API can be used in conjunction with {@link #seek(Map) seek(offsets)} to rewind consumption of data. |
| * @param partitions The list of partitions to return the last committed offset for |
| * @return The list of offsets committed on the last {@link #commit(boolean) commit(sync)} |
| */ |
| @Override |
| public Map<TopicPartition, Long> committed(Collection<TopicPartition> partitions) { |
| // TODO Auto-generated method stub |
| throw new UnsupportedOperationException(); |
| } |
| |
| /** |
| * Fetches offsets before a certain timestamp. Note that the offsets returned are approximately computed and do not correspond to the exact |
| * message at the given timestamp. As such, if the consumer is rewound to offsets returned by this API, there may be duplicate messages |
| * returned by the consumer. |
| * @param partitions The list of partitions for which the offsets are returned |
| * @param timestamp The unix timestamp. Value -1 indicates earliest available timestamp. Value -2 indicates latest available timestamp. |
| * @return The offsets per partition before the specified timestamp. |
| */ |
| public Map<TopicPartition, Long> offsetsBeforeTime(long timestamp, Collection<TopicPartition> partitions) { |
| return null; |
| } |
| |
| @Override |
| public Map<MetricName, ? extends Metric> metrics() { |
| return Collections.unmodifiableMap(this.metrics.metrics()); |
| } |
| |
| @Override |
| public void close() { |
| log.trace("Closing the Kafka consumer."); |
| subscribedTopics.clear(); |
| subscribedPartitions.clear(); |
| this.metrics.close(); |
| log.debug("The Kafka consumer has closed."); |
| } |
| } |
