| /** |
| * 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 static org.apache.kafka.common.config.ConfigDef.Range.atLeast; |
| |
| import java.util.Map; |
| |
| import org.apache.kafka.common.config.AbstractConfig; |
| import org.apache.kafka.common.config.ConfigDef; |
| import org.apache.kafka.common.config.ConfigDef.Importance; |
| import org.apache.kafka.common.config.ConfigDef.Type; |
| |
| /** |
| * The consumer configuration keys |
| */ |
| public class ConsumerConfig extends AbstractConfig { |
| private static final ConfigDef config; |
| |
| /** |
| * The identifier of the group this consumer belongs to. This is required if the consumer uses either the |
| * group management functionality by using {@link Consumer#subscribe(String...) subscribe(topics)}. This is also required |
| * if the consumer uses the default Kafka based offset management strategy. |
| */ |
| public static final String GROUP_ID_CONFIG = "group.id"; |
| |
| /** |
| * The timeout after which, if the {@link Consumer#poll(long) poll(timeout)} is not invoked, the consumer is |
| * marked dead and a rebalance operation is triggered for the group identified by {@link #GROUP_ID_CONFIG}. Relevant |
| * if the consumer uses the group management functionality by invoking {@link Consumer#subscribe(String...) subscribe(topics)} |
| */ |
| public static final String SESSION_TIMEOUT_MS = "session.timeout.ms"; |
| |
| /** |
| * The number of times a consumer sends a heartbeat to the co-ordinator broker within a {@link #SESSION_TIMEOUT_MS} time window. |
| * This frequency affects the latency of a rebalance operation since the co-ordinator broker notifies a consumer of a rebalance |
| * in the heartbeat response. Relevant if the consumer uses the group management functionality by invoking |
| * {@link Consumer#subscribe(String...) subscribe(topics)} |
| */ |
| public static final String HEARTBEAT_FREQUENCY = "heartbeat.frequency"; |
| |
| /** |
| * A list of URLs to use for establishing the initial connection to the cluster. This list should be in the form |
| * <code>host1:port1,host2:port2,...</code>. These urls are just used for the initial connection to discover the |
| * full cluster membership (which may change dynamically) so this list need not contain the full set of servers (you |
| * may want more than one, though, in case a server is down). |
| */ |
| public static final String BOOTSTRAP_SERVERS_CONFIG = "bootstrap.servers"; |
| |
| /** |
| * If true, periodically commit to Kafka the offsets of messages already returned by the consumer. This committed |
| * offset will be used when the process fails as the position from which the consumption will begin. |
| */ |
| public static final String ENABLE_AUTO_COMMIT_CONFIG = "enable.auto.commit"; |
| |
| /** |
| * The friendly name of the partition assignment strategy that the server will use to distribute partition ownership |
| * amongst consumer instances when group management is used |
| */ |
| public static final String PARTITION_ASSIGNMENT_STRATEGY = "partition.assignment.strategy"; |
| |
| /** |
| * The frequency in milliseconds that the consumer offsets are committed to Kafka. Relevant if {@link #ENABLE_AUTO_COMMIT_CONFIG} |
| * is turned on. |
| */ |
| public static final String AUTO_COMMIT_INTERVAL_MS_CONFIG = "auto.commit.interval.ms"; |
| |
| /** |
| * What to do when there is no initial offset in Kafka or if an offset is out of range: |
| * <ul> |
| * <li> smallest: automatically reset the offset to the smallest offset |
| * <li> largest: automatically reset the offset to the largest offset |
| * <li> disable: throw exception to the consumer if no previous offset is found for the consumer's group |
| * <li> anything else: throw exception to the consumer. |
| * </ul> |
| */ |
| public static final String AUTO_OFFSET_RESET_CONFIG = "auto.offset.reset"; |
| |
| /** |
| * The minimum amount of data the server should return for a fetch request. If insufficient data is available the |
| * request will wait for that much data to accumulate before answering the request. |
| */ |
| public static final String FETCH_MIN_BYTES_CONFIG = "fetch.min.bytes"; |
| |
| /** |
| * The maximum amount of time the server will block before answering the fetch request if there isn't sufficient |
| * data to immediately satisfy {@link #FETCH_MIN_BYTES_CONFIG}. This should be less than or equal to the timeout used in |
| * {@link KafkaConsumer#poll(long) poll(timeout)} |
| */ |
| public static final String FETCH_MAX_WAIT_MS_CONFIG = "fetch.max.wait.ms"; |
| |
| /** |
| * The maximum amount of time to block waiting to fetch metadata about a topic the first time a record is received |
| * from that topic. The consumer will throw a TimeoutException if it could not successfully fetch metadata within |
| * this timeout. |
| */ |
| public static final String METADATA_FETCH_TIMEOUT_CONFIG = "metadata.fetch.timeout.ms"; |
| |
| /** |
| * The total memory used by the consumer to buffer records received from the server. This config is meant to control |
| * the consumer's memory usage, so it is the size of the global fetch buffer that will be shared across all partitions. |
| */ |
| public static final String TOTAL_BUFFER_MEMORY_CONFIG = "total.memory.bytes"; |
| |
| /** |
| * The minimum amount of memory that should be used to fetch at least one message for a partition. This puts a lower |
| * bound on the consumer's memory utilization when there is at least one message for a partition available on the server. |
| * This size must be at least as large as the maximum message size the server allows or else it is possible for the producer |
| * to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large |
| * message on a certain partition. |
| */ |
| public static final String FETCH_BUFFER_CONFIG = "fetch.buffer.bytes"; |
| |
| /** |
| * The id string to pass to the server when making requests. The purpose of this is to be able to track the source |
| * of requests beyond just ip/port by allowing a logical application name to be included. |
| */ |
| public static final String CLIENT_ID_CONFIG = "client.id"; |
| |
| /** |
| * The size of the TCP send buffer to use when fetching data |
| */ |
| public static final String SOCKET_RECEIVE_BUFFER_CONFIG = "socket.receive.buffer.bytes"; |
| |
| /** |
| * The amount of time to wait before attempting to reconnect to a given host. This avoids repeatedly connecting to a |
| * host in a tight loop. This backoff applies to all requests sent by the consumer to the broker. |
| */ |
| public static final String RECONNECT_BACKOFF_MS_CONFIG = "reconnect.backoff.ms"; |
| |
| /** <code>metrics.sample.window.ms</code> */ |
| public static final String METRICS_SAMPLE_WINDOW_MS_CONFIG = "metrics.sample.window.ms"; |
| private static final String METRICS_SAMPLE_WINDOW_MS_DOC = "The metrics system maintains a configurable number of samples over a fixed window size. This configuration " + "controls the size of the window. For example we might maintain two samples each measured over a 30 second period. " |
| + "When a window expires we erase and overwrite the oldest window."; |
| |
| /** <code>metrics.num.samples</code> */ |
| public static final String METRICS_NUM_SAMPLES_CONFIG = "metrics.num.samples"; |
| private static final String METRICS_NUM_SAMPLES_DOC = "The number of samples maintained to compute metrics."; |
| |
| /** <code>metric.reporters</code> */ |
| public static final String METRIC_REPORTER_CLASSES_CONFIG = "metric.reporters"; |
| private static final String METRIC_REPORTER_CLASSES_DOC = "A list of classes to use as metrics reporters. Implementing the <code>MetricReporter</code> interface allows " + "plugging in classes that will be notified of new metric creation. The JmxReporter is always included to register JMX statistics."; |
| |
| /** <code>key.deserializer</code> */ |
| public static final String KEY_DESERIALIZER_CLASS_CONFIG = "key.deserializer"; |
| private static final String KEY_DESERIALIZER_CLASS_DOC = "Deserializer class for key that implements the <code>Deserializer</code> interface."; |
| |
| /** <code>value.deserializer</code> */ |
| public static final String VALUE_DESERIALIZER_CLASS_CONFIG = "value.deserializer"; |
| private static final String VALUE_DESERIALIZER_CLASS_DOC = "Deserializer class for value that implements the <code>Deserializer</code> interface."; |
| |
| static { |
| /* TODO: add config docs */ |
| config = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, Importance.HIGH, "blah blah") |
| .define(GROUP_ID_CONFIG, Type.STRING, Importance.HIGH, "blah blah") |
| .define(SESSION_TIMEOUT_MS, Type.LONG, 1000, Importance.HIGH, "blah blah") |
| .define(HEARTBEAT_FREQUENCY, Type.INT, 3, Importance.MEDIUM, "blah blah") |
| .define(PARTITION_ASSIGNMENT_STRATEGY, Type.STRING, Importance.MEDIUM, "blah blah") |
| .define(METADATA_FETCH_TIMEOUT_CONFIG, Type.LONG, 60 * 1000, atLeast(0), Importance.MEDIUM, "blah blah") |
| .define(ENABLE_AUTO_COMMIT_CONFIG, Type.BOOLEAN, true, Importance.MEDIUM, "blah blah") |
| .define(AUTO_COMMIT_INTERVAL_MS_CONFIG, Type.LONG, 5000, atLeast(0), Importance.LOW, "blah blah") |
| .define(CLIENT_ID_CONFIG, Type.STRING, "", Importance.LOW, "blah blah") |
| .define(TOTAL_BUFFER_MEMORY_CONFIG, Type.LONG, 32 * 1024 * 1024L, atLeast(0L), Importance.LOW, "blah blah") |
| .define(FETCH_BUFFER_CONFIG, Type.INT, 1 * 1024 * 1024, atLeast(0), Importance.HIGH, "blah blah") |
| .define(SOCKET_RECEIVE_BUFFER_CONFIG, Type.INT, 128 * 1024, atLeast(0), Importance.LOW, "blah blah") |
| .define(FETCH_MIN_BYTES_CONFIG, Type.LONG, 1024, atLeast(0), Importance.HIGH, "blah blah") |
| .define(FETCH_MAX_WAIT_MS_CONFIG, Type.LONG, 500, atLeast(0), Importance.LOW, "blah blah") |
| .define(RECONNECT_BACKOFF_MS_CONFIG, Type.LONG, 10L, atLeast(0L), Importance.LOW, "blah blah") |
| .define(AUTO_OFFSET_RESET_CONFIG, Type.STRING, "largest", Importance.MEDIUM, "blah blah") |
| .define(METRICS_SAMPLE_WINDOW_MS_CONFIG, |
| Type.LONG, |
| 30000, |
| atLeast(0), |
| Importance.LOW, |
| METRICS_SAMPLE_WINDOW_MS_DOC) |
| .define(METRICS_NUM_SAMPLES_CONFIG, Type.INT, 2, atLeast(1), Importance.LOW, METRICS_NUM_SAMPLES_DOC) |
| .define(METRIC_REPORTER_CLASSES_CONFIG, Type.LIST, "", Importance.LOW, METRIC_REPORTER_CLASSES_DOC) |
| .define(KEY_DESERIALIZER_CLASS_CONFIG, Type.CLASS, Importance.HIGH, KEY_DESERIALIZER_CLASS_DOC) |
| .define(VALUE_DESERIALIZER_CLASS_CONFIG, Type.CLASS, Importance.HIGH, VALUE_DESERIALIZER_CLASS_DOC); |
| |
| } |
| |
| ConsumerConfig(Map<? extends Object, ? extends Object> props) { |
| super(config, props); |
| } |
| |
| } |
