diff --git a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigDefs.java b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigDefs.java new file mode 100644 index 0000000000000..19e1d0ea9db92 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigDefs.java @@ -0,0 +1,187 @@ +/* + * 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; + +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigDef.ConfigKey; +import org.apache.kafka.common.metrics.Sensor; + +import java.util.Collections; + +import static org.apache.kafka.common.config.ConfigDef.Importance; +import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; +import static org.apache.kafka.common.config.ConfigDef.Range.between; +import static org.apache.kafka.common.config.ConfigDef.Type; +import static org.apache.kafka.common.config.ConfigDef.ValidString.in; + +public class CommonClientConfigDefs { + + public static ConfigKey sendBufferBytes(Integer defaultValue) { + return new ConfigKey( + CommonClientConfigs.SEND_BUFFER_CONFIG, + Type.INT, + defaultValue, + atLeast(-1), + Importance.MEDIUM, + CommonClientConfigs.SEND_BUFFER_DOC); + } + + public static ConfigKey receiveBufferBytes(Integer defaultValue) { + return new ConfigKey( + CommonClientConfigs.RECEIVE_BUFFER_CONFIG, + Type.INT, + defaultValue, + atLeast(-1), + Importance.MEDIUM, + CommonClientConfigs.RECEIVE_BUFFER_DOC); + } + + public static ConfigKey reconnectBackoffMs(Long defaultValue) { + return new ConfigKey( + CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG, + Type.LONG, + defaultValue, + atLeast(0L), + Importance.LOW, + CommonClientConfigs.RECONNECT_BACKOFF_MS_DOC); + } + + public static ConfigKey reconnectBackoffMaxMs(Long defaultValue) { + return new ConfigKey( + CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_CONFIG, + Type.LONG, + defaultValue, + atLeast(0L), + Importance.LOW, + CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_DOC); + } + + public static ConfigKey retryBackoffMs(Long defaultValue) { + return new ConfigKey( + CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG, + Type.LONG, + defaultValue, + atLeast(0L), + Importance.LOW, + CommonClientConfigs.RETRY_BACKOFF_MS_DOC); + } + + public static ConfigKey requestTimeoutMs(Integer defaultValue, String doc) { + return new ConfigKey( + CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG, + Type.INT, + defaultValue, + atLeast(0), + Importance.MEDIUM, + doc); + } + + public static ConfigKey connectionsMaxIdleMs(Long defaultValue) { + return new ConfigKey( + CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG, + Type.LONG, + 5 * 60 * 1000, + Importance.MEDIUM, + CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC); + } + + public static ConfigKey metadataMaxAge(Long defaultValue) { + return new ConfigKey( + CommonClientConfigs.METADATA_MAX_AGE_CONFIG, + Type.LONG, + defaultValue, + atLeast(0), + Importance.LOW, + CommonClientConfigs.METADATA_MAX_AGE_DOC); + } + + public static ConfigKey metricsSampleWindowMs(Long defaultValue) { + return new ConfigKey( + CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG, + Type.LONG, + defaultValue, + atLeast(0), + Importance.LOW, + CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_DOC); + } + + public static ConfigKey metricReporterClasses() { + return new ConfigKey( + CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, + Type.LIST, + Collections.emptyList(), + Importance.LOW, + CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC); + } + + public static ConfigKey metricsRecordingLevel() { + return new ConfigKey( + CommonClientConfigs.METRICS_RECORDING_LEVEL_CONFIG, + Type.STRING, + Sensor.RecordingLevel.INFO.toString(), + in(Sensor.RecordingLevel.INFO.toString(), Sensor.RecordingLevel.DEBUG.toString()), + Importance.LOW, + CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC); + } + + public static ConfigKey metricsNumSamplesConfig(Integer defaultValue) { + return new ConfigKey( + CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG, + Type.INT, + defaultValue, + atLeast(1), + Importance.LOW, + CommonClientConfigs.METRICS_NUM_SAMPLES_DOC); + } + + public static ConfigKey bootstrapServers(Object defaultValue, String doc) { + return new ConfigKey( + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, + Type.LIST, + defaultValue, + new ConfigDef.NonNullValidator(), + Importance.HIGH, + doc); + } + + public static ConfigKey retries(Integer defaultValue, String doc) { + return new ConfigKey( + CommonClientConfigs.RETRIES_CONFIG, + Type.INT, + defaultValue, + between(0, Integer.MAX_VALUE), Importance.HIGH, + doc); + } + + public static ConfigKey securityProtocol() { + return new ConfigKey( + CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, + Type.STRING, + CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL, + Importance.MEDIUM, + CommonClientConfigs.SECURITY_PROTOCOL_DOC); + } + + public static ConfigKey clientId(String doc) { + return new ConfigKey( + CommonClientConfigs.CLIENT_ID_CONFIG, + Type.STRING, + "", + Importance.LOW, + doc); + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java index b5ca15a7a6fdd..1f191fbdbb45a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java @@ -17,19 +17,14 @@ package org.apache.kafka.clients.admin; +import org.apache.kafka.clients.CommonClientConfigDefs; import org.apache.kafka.clients.CommonClientConfigs; 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; -import org.apache.kafka.common.metrics.Sensor; import java.util.Map; import java.util.Set; -import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; -import static org.apache.kafka.common.config.ConfigDef.ValidString.in; - /** * The AdminClient configuration class, which also contains constants for configuration entry names. */ @@ -40,19 +35,16 @@ public class AdminClientConfig extends AbstractConfig { * bootstrap.servers */ public static final String BOOTSTRAP_SERVERS_CONFIG = CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG; - private static final String BOOTSTRAP_SERVERS_DOC = CommonClientConfigs.BOOTSTRAP_SERVERS_DOC; /** * reconnect.backoff.ms */ public static final String RECONNECT_BACKOFF_MS_CONFIG = CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG; - private static final String RECONNECT_BACKOFF_MS_DOC = CommonClientConfigs.RECONNECT_BACKOFF_MS_DOC; /** * reconnect.backoff.max.ms */ public static final String RECONNECT_BACKOFF_MAX_MS_CONFIG = CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_CONFIG; - private static final String RECONNECT_BACKOFF_MAX_MS_DOC = CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_DOC; /** * retry.backoff.ms @@ -64,106 +56,52 @@ public class AdminClientConfig extends AbstractConfig { /** connections.max.idle.ms */ public static final String CONNECTIONS_MAX_IDLE_MS_CONFIG = CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG; - private static final String CONNECTIONS_MAX_IDLE_MS_DOC = CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC; /** request.timeout.ms */ public static final String REQUEST_TIMEOUT_MS_CONFIG = CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG; - private static final String REQUEST_TIMEOUT_MS_DOC = CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC; public static final String CLIENT_ID_CONFIG = CommonClientConfigs.CLIENT_ID_CONFIG; private static final String CLIENT_ID_DOC = CommonClientConfigs.CLIENT_ID_DOC; public static final String METADATA_MAX_AGE_CONFIG = CommonClientConfigs.METADATA_MAX_AGE_CONFIG; - private static final String METADATA_MAX_AGE_DOC = CommonClientConfigs.METADATA_MAX_AGE_DOC; public static final String SEND_BUFFER_CONFIG = CommonClientConfigs.SEND_BUFFER_CONFIG; - private static final String SEND_BUFFER_DOC = CommonClientConfigs.SEND_BUFFER_DOC; public static final String RECEIVE_BUFFER_CONFIG = CommonClientConfigs.RECEIVE_BUFFER_CONFIG; - private static final String RECEIVE_BUFFER_DOC = CommonClientConfigs.RECEIVE_BUFFER_DOC; public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG; - private static final String METRIC_REPORTER_CLASSES_DOC = CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC; public static final String METRICS_NUM_SAMPLES_CONFIG = CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG; - private static final String METRICS_NUM_SAMPLES_DOC = CommonClientConfigs.METRICS_NUM_SAMPLES_DOC; public static final String METRICS_SAMPLE_WINDOW_MS_CONFIG = CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG; - private static final String METRICS_SAMPLE_WINDOW_MS_DOC = CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_DOC; public static final String METRICS_RECORDING_LEVEL_CONFIG = CommonClientConfigs.METRICS_RECORDING_LEVEL_CONFIG; public static final String SECURITY_PROTOCOL_CONFIG = CommonClientConfigs.SECURITY_PROTOCOL_CONFIG; public static final String DEFAULT_SECURITY_PROTOCOL = CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL; - private static final String SECURITY_PROTOCOL_DOC = CommonClientConfigs.SECURITY_PROTOCOL_DOC; - private static final String METRICS_RECORDING_LEVEL_DOC = CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC; public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG; static { - CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, - Type.LIST, - Importance.HIGH, - BOOTSTRAP_SERVERS_DOC) - .define(CLIENT_ID_CONFIG, Type.STRING, "", Importance.MEDIUM, CLIENT_ID_DOC) - .define(METADATA_MAX_AGE_CONFIG, Type.LONG, 5 * 60 * 1000, atLeast(0), Importance.LOW, METADATA_MAX_AGE_DOC) - .define(SEND_BUFFER_CONFIG, Type.INT, 128 * 1024, atLeast(-1), Importance.MEDIUM, SEND_BUFFER_DOC) - .define(RECEIVE_BUFFER_CONFIG, Type.INT, 64 * 1024, atLeast(-1), Importance.MEDIUM, RECEIVE_BUFFER_DOC) - .define(RECONNECT_BACKOFF_MS_CONFIG, - Type.LONG, - 50L, - atLeast(0L), - Importance.LOW, - RECONNECT_BACKOFF_MS_DOC) - .define(RECONNECT_BACKOFF_MAX_MS_CONFIG, - Type.LONG, - 1000L, - atLeast(0L), - Importance.LOW, - RECONNECT_BACKOFF_MAX_MS_DOC) - .define(RETRY_BACKOFF_MS_CONFIG, - Type.LONG, - 100L, - atLeast(0L), - Importance.LOW, - RETRY_BACKOFF_MS_DOC) - .define(REQUEST_TIMEOUT_MS_CONFIG, - Type.INT, - 120000, - atLeast(0), - Importance.MEDIUM, - REQUEST_TIMEOUT_MS_DOC) - .define(CONNECTIONS_MAX_IDLE_MS_CONFIG, - Type.LONG, - 5 * 60 * 1000, - Importance.MEDIUM, - CONNECTIONS_MAX_IDLE_MS_DOC) - .define(RETRIES_CONFIG, - Type.INT, - 5, - atLeast(0), - Importance.LOW, - CommonClientConfigs.RETRIES_DOC) - .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(METRICS_RECORDING_LEVEL_CONFIG, - Type.STRING, - Sensor.RecordingLevel.INFO.toString(), - in(Sensor.RecordingLevel.INFO.toString(), Sensor.RecordingLevel.DEBUG.toString()), - Importance.LOW, - METRICS_RECORDING_LEVEL_DOC) + CONFIG = new ConfigDef().define(CommonClientConfigDefs.bootstrapServers(ConfigDef.NO_DEFAULT_VALUE, CommonClientConfigs.BOOTSTRAP_SERVERS_DOC)) + .define(CommonClientConfigDefs.clientId(CLIENT_ID_DOC)) + .define(CommonClientConfigDefs.metadataMaxAge(5 * 60 * 1000L)) + .define(CommonClientConfigDefs.sendBufferBytes(128 * 1024)) + .define(CommonClientConfigDefs.receiveBufferBytes(64 * 1024)) + .define(CommonClientConfigDefs.reconnectBackoffMs(50L)) + .define(CommonClientConfigDefs.reconnectBackoffMaxMs(1000L)) + .define(CommonClientConfigDefs.retryBackoffMs(100L)) + .define(CommonClientConfigDefs.requestTimeoutMs(120000, CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC)) + .define(CommonClientConfigDefs.connectionsMaxIdleMs(5 * 60 * 1000L)) + + .define(CommonClientConfigDefs.retries(5, CommonClientConfigs.RETRIES_DOC)) + + .define(CommonClientConfigDefs.metricsSampleWindowMs(30_000L)) + .define(CommonClientConfigDefs.metricsNumSamplesConfig(2)) + .define(CommonClientConfigDefs.metricReporterClasses()) + .define(CommonClientConfigDefs.metricsRecordingLevel()) // security support - .define(SECURITY_PROTOCOL_CONFIG, - Type.STRING, - DEFAULT_SECURITY_PROTOCOL, - Importance.MEDIUM, - SECURITY_PROTOCOL_DOC) + .define(CommonClientConfigDefs.securityProtocol()) .withClientSslSupport() .withClientSaslSupport(); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java index 72e496cbd4697..a38f8c18042ee 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java @@ -16,12 +16,12 @@ */ package org.apache.kafka.clients.consumer; +import org.apache.kafka.clients.CommonClientConfigDefs; import org.apache.kafka.clients.CommonClientConfigs; 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; -import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.requests.IsolationLevel; import org.apache.kafka.common.serialization.Deserializer; @@ -32,6 +32,7 @@ import java.util.Properties; import java.util.Set; +import static org.apache.kafka.clients.CommonClientConfigs.CLIENT_ID_DOC; import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; import static org.apache.kafka.common.config.ConfigDef.ValidString.in; @@ -216,7 +217,6 @@ public class ConsumerConfig extends AbstractConfig { /** request.timeout.ms */ public static final String REQUEST_TIMEOUT_MS_CONFIG = CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG; - private static final String REQUEST_TIMEOUT_MS_DOC = CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC; /** interceptor.classes */ public static final String INTERCEPTOR_CLASSES_CONFIG = "interceptor.classes"; @@ -255,12 +255,7 @@ public class ConsumerConfig extends AbstractConfig { public static final String DEFAULT_ISOLATION_LEVEL = IsolationLevel.READ_UNCOMMITTED.toString().toLowerCase(Locale.ROOT); static { - CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, - Type.LIST, - Collections.emptyList(), - new ConfigDef.NonNullValidator(), - Importance.HIGH, - CommonClientConfigs.BOOTSTRAP_SERVERS_DOC) + CONFIG = new ConfigDef().define(CommonClientConfigDefs.bootstrapServers(ConfigDef.NO_DEFAULT_VALUE, CommonClientConfigs.BOOTSTRAP_SERVERS_DOC)) .define(GROUP_ID_CONFIG, Type.STRING, "", Importance.HIGH, GROUP_ID_DOC) .define(SESSION_TIMEOUT_MS_CONFIG, Type.INT, @@ -278,12 +273,7 @@ public class ConsumerConfig extends AbstractConfig { new ConfigDef.NonNullValidator(), Importance.MEDIUM, PARTITION_ASSIGNMENT_STRATEGY_DOC) - .define(METADATA_MAX_AGE_CONFIG, - Type.LONG, - 5 * 60 * 1000, - atLeast(0), - Importance.LOW, - CommonClientConfigs.METADATA_MAX_AGE_DOC) + .define(CommonClientConfigDefs.metadataMaxAge(5 * 60 * 1000L)) .define(ENABLE_AUTO_COMMIT_CONFIG, Type.BOOLEAN, true, @@ -295,29 +285,15 @@ public class ConsumerConfig extends AbstractConfig { atLeast(0), Importance.LOW, AUTO_COMMIT_INTERVAL_MS_DOC) - .define(CLIENT_ID_CONFIG, - Type.STRING, - "", - Importance.LOW, - CommonClientConfigs.CLIENT_ID_DOC) + .define(CommonClientConfigDefs.clientId(CLIENT_ID_DOC)) .define(MAX_PARTITION_FETCH_BYTES_CONFIG, Type.INT, DEFAULT_MAX_PARTITION_FETCH_BYTES, atLeast(0), Importance.HIGH, MAX_PARTITION_FETCH_BYTES_DOC) - .define(SEND_BUFFER_CONFIG, - Type.INT, - 128 * 1024, - atLeast(-1), - Importance.MEDIUM, - CommonClientConfigs.SEND_BUFFER_DOC) - .define(RECEIVE_BUFFER_CONFIG, - Type.INT, - 64 * 1024, - atLeast(-1), - Importance.MEDIUM, - CommonClientConfigs.RECEIVE_BUFFER_DOC) + .define(CommonClientConfigDefs.sendBufferBytes(128 * 1024)) + .define(CommonClientConfigDefs.receiveBufferBytes(64 * 1024)) .define(FETCH_MIN_BYTES_CONFIG, Type.INT, 1, @@ -336,24 +312,9 @@ public class ConsumerConfig extends AbstractConfig { atLeast(0), Importance.LOW, FETCH_MAX_WAIT_MS_DOC) - .define(RECONNECT_BACKOFF_MS_CONFIG, - Type.LONG, - 50L, - atLeast(0L), - Importance.LOW, - CommonClientConfigs.RECONNECT_BACKOFF_MS_DOC) - .define(RECONNECT_BACKOFF_MAX_MS_CONFIG, - Type.LONG, - 1000L, - atLeast(0L), - Importance.LOW, - CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_DOC) - .define(RETRY_BACKOFF_MS_CONFIG, - Type.LONG, - 100L, - atLeast(0L), - Importance.LOW, - CommonClientConfigs.RETRY_BACKOFF_MS_DOC) + .define(CommonClientConfigDefs.reconnectBackoffMs(50L)) + .define(CommonClientConfigDefs.reconnectBackoffMaxMs(1000L)) + .define(CommonClientConfigDefs.retryBackoffMs(100L)) .define(AUTO_OFFSET_RESET_CONFIG, Type.STRING, "latest", @@ -365,30 +326,10 @@ public class ConsumerConfig extends AbstractConfig { true, Importance.LOW, CHECK_CRCS_DOC) - .define(METRICS_SAMPLE_WINDOW_MS_CONFIG, - Type.LONG, - 30000, - atLeast(0), - Importance.LOW, - CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_DOC) - .define(METRICS_NUM_SAMPLES_CONFIG, - Type.INT, - 2, - atLeast(1), - Importance.LOW, - CommonClientConfigs.METRICS_NUM_SAMPLES_DOC) - .define(METRICS_RECORDING_LEVEL_CONFIG, - Type.STRING, - Sensor.RecordingLevel.INFO.toString(), - in(Sensor.RecordingLevel.INFO.toString(), Sensor.RecordingLevel.DEBUG.toString()), - Importance.LOW, - CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC) - .define(METRIC_REPORTER_CLASSES_CONFIG, - Type.LIST, - Collections.emptyList(), - new ConfigDef.NonNullValidator(), - Importance.LOW, - CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC) + .define(CommonClientConfigDefs.metricsSampleWindowMs(30_000L)) + .define(CommonClientConfigDefs.metricsNumSamplesConfig(2)) + .define(CommonClientConfigDefs.metricsRecordingLevel()) + .define(CommonClientConfigDefs.metricReporterClasses()) .define(KEY_DESERIALIZER_CLASS_CONFIG, Type.CLASS, Importance.HIGH, @@ -397,18 +338,9 @@ public class ConsumerConfig extends AbstractConfig { Type.CLASS, Importance.HIGH, VALUE_DESERIALIZER_CLASS_DOC) - .define(REQUEST_TIMEOUT_MS_CONFIG, - Type.INT, - 305000, // chosen to be higher than the default of max.poll.interval.ms - atLeast(0), - Importance.MEDIUM, - REQUEST_TIMEOUT_MS_DOC) + .define(CommonClientConfigDefs.requestTimeoutMs(305000, CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC)) /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */ - .define(CONNECTIONS_MAX_IDLE_MS_CONFIG, - Type.LONG, - 9 * 60 * 1000, - Importance.MEDIUM, - CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC) + .define(CommonClientConfigDefs.connectionsMaxIdleMs(9 * 60 * 1000L)) .define(INTERCEPTOR_CLASSES_CONFIG, Type.LIST, Collections.emptyList(), @@ -443,11 +375,8 @@ public class ConsumerConfig extends AbstractConfig { Importance.MEDIUM, ISOLATION_LEVEL_DOC) // security support - .define(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, - Type.STRING, - CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL, - Importance.MEDIUM, - CommonClientConfigs.SECURITY_PROTOCOL_DOC) + .define(CommonClientConfigDefs.securityProtocol()) + .withClientSslSupport() .withClientSaslSupport(); diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java index becbef6d9f19d..72d2b50acf8f2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java @@ -16,13 +16,13 @@ */ package org.apache.kafka.clients.producer; +import org.apache.kafka.clients.CommonClientConfigDefs; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.producer.internals.DefaultPartitioner; 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; -import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.serialization.Serializer; import java.util.Collections; @@ -31,8 +31,8 @@ import java.util.Properties; import java.util.Set; +import static org.apache.kafka.clients.CommonClientConfigs.CLIENT_ID_DOC; import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; -import static org.apache.kafka.common.config.ConfigDef.Range.between; import static org.apache.kafka.common.config.ConfigDef.ValidString.in; /** @@ -53,7 +53,6 @@ public class ProducerConfig extends AbstractConfig { /** metadata.max.age.ms */ public static final String METADATA_MAX_AGE_CONFIG = CommonClientConfigs.METADATA_MAX_AGE_CONFIG; - private static final String METADATA_MAX_AGE_DOC = CommonClientConfigs.METADATA_MAX_AGE_DOC; /** batch.size */ public static final String BATCH_SIZE_CONFIG = "batch.size"; @@ -222,9 +221,9 @@ public class ProducerConfig extends AbstractConfig { "Note that transactions requires a cluster of at least three brokers by default what is the recommended setting for production; for development you can change this, by adjusting broker setting `transaction.state.log.replication.factor`."; static { - CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, Collections.emptyList(), new ConfigDef.NonNullValidator(), Importance.HIGH, CommonClientConfigs.BOOTSTRAP_SERVERS_DOC) + CONFIG = new ConfigDef().define(CommonClientConfigDefs.bootstrapServers(ConfigDef.NO_DEFAULT_VALUE, CommonClientConfigs.BOOTSTRAP_SERVERS_DOC)) .define(BUFFER_MEMORY_CONFIG, Type.LONG, 32 * 1024 * 1024L, atLeast(0L), Importance.HIGH, BUFFER_MEMORY_DOC) - .define(RETRIES_CONFIG, Type.INT, 0, between(0, Integer.MAX_VALUE), Importance.HIGH, RETRIES_DOC) + .define(CommonClientConfigDefs.retries(0, RETRIES_DOC)) .define(ACKS_CONFIG, Type.STRING, "1", @@ -234,8 +233,8 @@ public class ProducerConfig extends AbstractConfig { .define(COMPRESSION_TYPE_CONFIG, Type.STRING, "none", Importance.HIGH, COMPRESSION_TYPE_DOC) .define(BATCH_SIZE_CONFIG, Type.INT, 16384, atLeast(0), Importance.MEDIUM, BATCH_SIZE_DOC) .define(LINGER_MS_CONFIG, Type.LONG, 0, atLeast(0L), Importance.MEDIUM, LINGER_MS_DOC) - .define(CLIENT_ID_CONFIG, Type.STRING, "", Importance.MEDIUM, CommonClientConfigs.CLIENT_ID_DOC) - .define(SEND_BUFFER_CONFIG, Type.INT, 128 * 1024, atLeast(-1), Importance.MEDIUM, CommonClientConfigs.SEND_BUFFER_DOC) + .define(CommonClientConfigDefs.clientId(CLIENT_ID_DOC)) + .define(CommonClientConfigDefs.sendBufferBytes(128 * 1024)) .define(RECEIVE_BUFFER_CONFIG, Type.INT, 32 * 1024, atLeast(-1), Importance.MEDIUM, CommonClientConfigs.RECEIVE_BUFFER_DOC) .define(MAX_REQUEST_SIZE_CONFIG, Type.INT, @@ -243,41 +242,21 @@ public class ProducerConfig extends AbstractConfig { atLeast(0), Importance.MEDIUM, MAX_REQUEST_SIZE_DOC) - .define(RECONNECT_BACKOFF_MS_CONFIG, Type.LONG, 50L, atLeast(0L), Importance.LOW, CommonClientConfigs.RECONNECT_BACKOFF_MS_DOC) - .define(RECONNECT_BACKOFF_MAX_MS_CONFIG, Type.LONG, 1000L, atLeast(0L), Importance.LOW, CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_DOC) - .define(RETRY_BACKOFF_MS_CONFIG, Type.LONG, 100L, atLeast(0L), Importance.LOW, CommonClientConfigs.RETRY_BACKOFF_MS_DOC) + .define(CommonClientConfigDefs.reconnectBackoffMs(50L)) + .define(CommonClientConfigDefs.reconnectBackoffMaxMs(1000L)) + .define(CommonClientConfigDefs.retryBackoffMs(100L)) .define(MAX_BLOCK_MS_CONFIG, Type.LONG, 60 * 1000, atLeast(0), Importance.MEDIUM, MAX_BLOCK_MS_DOC) - .define(REQUEST_TIMEOUT_MS_CONFIG, - Type.INT, - 30 * 1000, - atLeast(0), - Importance.MEDIUM, - REQUEST_TIMEOUT_MS_DOC) - .define(METADATA_MAX_AGE_CONFIG, Type.LONG, 5 * 60 * 1000, atLeast(0), Importance.LOW, METADATA_MAX_AGE_DOC) - .define(METRICS_SAMPLE_WINDOW_MS_CONFIG, - Type.LONG, - 30000, - atLeast(0), - Importance.LOW, - CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_DOC) - .define(METRICS_NUM_SAMPLES_CONFIG, Type.INT, 2, atLeast(1), Importance.LOW, CommonClientConfigs.METRICS_NUM_SAMPLES_DOC) - .define(METRICS_RECORDING_LEVEL_CONFIG, - Type.STRING, - Sensor.RecordingLevel.INFO.toString(), - in(Sensor.RecordingLevel.INFO.toString(), Sensor.RecordingLevel.DEBUG.toString()), - Importance.LOW, - CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC) - .define(METRIC_REPORTER_CLASSES_CONFIG, - Type.LIST, - Collections.emptyList(), - new ConfigDef.NonNullValidator(), - Importance.LOW, - CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC) + .define(CommonClientConfigDefs.requestTimeoutMs(30 * 1000, REQUEST_TIMEOUT_MS_DOC)) + .define(CommonClientConfigDefs.metadataMaxAge(5 * 60 * 1000L)) + .define(CommonClientConfigDefs.metricsSampleWindowMs(30_000L)) + .define(CommonClientConfigDefs.metricsNumSamplesConfig(2)) + .define(CommonClientConfigDefs.metricsRecordingLevel()) + .define(CommonClientConfigDefs.metricReporterClasses()) .define(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, Type.INT, 5, @@ -293,11 +272,7 @@ public class ProducerConfig extends AbstractConfig { Importance.HIGH, VALUE_SERIALIZER_CLASS_DOC) /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */ - .define(CONNECTIONS_MAX_IDLE_MS_CONFIG, - Type.LONG, - 9 * 60 * 1000, - Importance.MEDIUM, - CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC) + .define(CommonClientConfigDefs.connectionsMaxIdleMs(9 * 60 * 1000L)) .define(PARTITIONER_CLASS_CONFIG, Type.CLASS, DefaultPartitioner.class, @@ -308,11 +283,7 @@ public class ProducerConfig extends AbstractConfig { new ConfigDef.NonNullValidator(), Importance.LOW, INTERCEPTOR_CLASSES_DOC) - .define(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, - Type.STRING, - CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL, - Importance.MEDIUM, - CommonClientConfigs.SECURITY_PROTOCOL_DOC) + .define(CommonClientConfigDefs.securityProtocol()) .withClientSslSupport() .withClientSaslSupport() .define(ENABLE_IDEMPOTENCE_CONFIG, diff --git a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java index a1c0e3be574db..80ba7b45d10a3 100644 --- a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java +++ b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java @@ -1053,6 +1053,16 @@ public ConfigKey(String name, Type type, Object defaultValue, Validator validato this.internalConfig = internalConfig; } + public ConfigKey(String name, Type type, Object defaultValue, Validator validator, Importance importance, String documentation) { + this(name, type, defaultValue, validator, importance, documentation, null, + -1, Width.NONE, name, Collections.emptyList(), null, false); + } + + public ConfigKey(String name, Type type, Object defaultValue, Importance importance, String documentation) { + this(name, type, defaultValue, null, importance, documentation, null, + -1, Width.NONE, name, Collections.emptyList(), null, false); + } + public boolean hasDefault() { return !NO_DEFAULT_VALUE.equals(this.defaultValue); } diff --git a/clients/src/test/java/org/apache/kafka/clients/CommonClientConfigsTest.java b/clients/src/test/java/org/apache/kafka/clients/CommonClientConfigsTest.java index 63a9312846001..68d65aafbd4a8 100644 --- a/clients/src/test/java/org/apache/kafka/clients/CommonClientConfigsTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/CommonClientConfigsTest.java @@ -25,7 +25,6 @@ import java.util.HashMap; import java.util.Map; -import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; import static org.junit.Assert.assertEquals; public class CommonClientConfigsTest { @@ -33,18 +32,8 @@ private static class TestConfig extends AbstractConfig { private static final ConfigDef CONFIG; static { CONFIG = new ConfigDef() - .define(CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG, - ConfigDef.Type.LONG, - 50L, - atLeast(0L), - ConfigDef.Importance.LOW, - "") - .define(CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_CONFIG, - ConfigDef.Type.LONG, - 1000L, - atLeast(0L), - ConfigDef.Importance.LOW, - ""); + .define(CommonClientConfigDefs.reconnectBackoffMs(50L)) + .define(CommonClientConfigDefs.reconnectBackoffMaxMs(1000L)); } @Override diff --git a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java index 071deed47e422..36ad611d6fcea 100644 --- a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java +++ b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java @@ -16,6 +16,8 @@ */ package org.apache.kafka.common.config; +import org.apache.kafka.clients.CommonClientConfigDefs; +import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.config.ConfigDef.Importance; import org.apache.kafka.common.config.ConfigDef.Type; @@ -220,23 +222,23 @@ public void testValuesWithPrefixAllOrNothing() { public void testUnused() { Properties props = new Properties(); String configValue = "org.apache.kafka.common.config.AbstractConfigTest$ConfiguredFakeMetricsReporter"; - props.put(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, configValue); + props.put(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, configValue); props.put(FakeMetricsReporterConfig.EXTRA_CONFIG, "my_value"); TestConfig config = new TestConfig(props); assertTrue("metric.extra_config should be marked unused before getConfiguredInstances is called", config.unused().contains(FakeMetricsReporterConfig.EXTRA_CONFIG)); - config.getConfiguredInstances(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class); + config.getConfiguredInstances(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class); assertTrue("All defined configurations should be marked as used", config.unused().isEmpty()); } private void testValidInputs(String configValue) { Properties props = new Properties(); - props.put(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, configValue); + props.put(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, configValue); TestConfig config = new TestConfig(props); try { - config.getConfiguredInstances(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class); + config.getConfiguredInstances(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class); } catch (ConfigException e) { fail("No exceptions are expected here, valid props are :" + props); } @@ -244,10 +246,10 @@ private void testValidInputs(String configValue) { private void testInvalidInputs(String configValue) { Properties props = new Properties(); - props.put(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, configValue); + props.put(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, configValue); TestConfig config = new TestConfig(props); try { - config.getConfiguredInstances(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class); + config.getConfiguredInstances(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class); fail("Expected a config exception due to invalid props :" + props); } catch (KafkaException e) { // this is good @@ -372,15 +374,9 @@ private static class TestConfig extends AbstractConfig { private static final ConfigDef CONFIG; - 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."; - static { - CONFIG = new ConfigDef().define(METRIC_REPORTER_CLASSES_CONFIG, - Type.LIST, - "", - Importance.LOW, - METRIC_REPORTER_CLASSES_DOC); + CONFIG = new ConfigDef() + .define(CommonClientConfigDefs.metricReporterClasses()); } public TestConfig(Map props) { diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java index b857b0e05658a..a0d356dde24aa 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java @@ -16,13 +16,13 @@ */ package org.apache.kafka.connect.runtime; +import org.apache.kafka.clients.CommonClientConfigDefs; import org.apache.kafka.clients.CommonClientConfigs; 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; import org.apache.kafka.common.config.internals.BrokerSecurityConfigs; -import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.connect.storage.SimpleHeaderConverter; import java.util.ArrayList; @@ -30,9 +30,6 @@ import java.util.List; import java.util.Map; -import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; -import static org.apache.kafka.common.config.ConfigDef.ValidString.in; - /** * Common base class providing configuration for Kafka Connect workers, whether standalone or distributed. */ @@ -172,11 +169,6 @@ public class WorkerConfig extends AbstractConfig { + "Examples: plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins," + "/opt/connectors"; - public static final String METRICS_SAMPLE_WINDOW_MS_CONFIG = CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG; - public static final String METRICS_NUM_SAMPLES_CONFIG = CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG; - public static final String METRICS_RECORDING_LEVEL_CONFIG = CommonClientConfigs.METRICS_RECORDING_LEVEL_CONFIG; - public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG; - /** * Get a basic ConfigDef for a WorkerConfig. This includes all the common settings. Subclasses can use this to * bootstrap their own ConfigDef. @@ -184,8 +176,7 @@ public class WorkerConfig extends AbstractConfig { */ protected static ConfigDef baseConfigDef() { return new ConfigDef() - .define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, BOOTSTRAP_SERVERS_DEFAULT, - Importance.HIGH, BOOTSTRAP_SERVERS_DOC) + .define(CommonClientConfigDefs.bootstrapServers(BOOTSTRAP_SERVERS_DEFAULT, BOOTSTRAP_SERVERS_DOC)) .define(KEY_CONVERTER_CLASS_CONFIG, Type.CLASS, Importance.HIGH, KEY_CONVERTER_CLASS_DOC) .define(VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS, @@ -218,20 +209,10 @@ protected static ConfigDef baseConfigDef() { null, Importance.LOW, PLUGIN_PATH_DOC) - .define(METRICS_SAMPLE_WINDOW_MS_CONFIG, Type.LONG, - 30000, atLeast(0), Importance.LOW, - CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_DOC) - .define(METRICS_NUM_SAMPLES_CONFIG, Type.INT, - 2, atLeast(1), Importance.LOW, - CommonClientConfigs.METRICS_NUM_SAMPLES_DOC) - .define(METRICS_RECORDING_LEVEL_CONFIG, Type.STRING, - Sensor.RecordingLevel.INFO.toString(), - in(Sensor.RecordingLevel.INFO.toString(), Sensor.RecordingLevel.DEBUG.toString()), - Importance.LOW, - CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC) - .define(METRIC_REPORTER_CLASSES_CONFIG, Type.LIST, - "", Importance.LOW, - CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC) + .define(CommonClientConfigDefs.metricsSampleWindowMs(30_000L)) + .define(CommonClientConfigDefs.metricsNumSamplesConfig(2)) + .define(CommonClientConfigDefs.metricsRecordingLevel()) + .define(CommonClientConfigDefs.metricReporterClasses()) .define(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, ConfigDef.Type.STRING, "none", ConfigDef.Importance.LOW, BrokerSecurityConfigs.SSL_CLIENT_AUTH_DOC) .define(HEADER_CONVERTER_CLASS_CONFIG, Type.CLASS, diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java index dc9017beeda75..f7a699555a5e0 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java @@ -16,12 +16,14 @@ */ package org.apache.kafka.connect.runtime.distributed; +import org.apache.kafka.clients.CommonClientConfigDefs; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.connect.runtime.WorkerConfig; import java.util.Map; +import static org.apache.kafka.clients.CommonClientConfigs.CLIENT_ID_DOC; import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; public class DistributedConfig extends WorkerConfig { @@ -153,67 +155,23 @@ public class DistributedConfig extends WorkerConfig { 3000, ConfigDef.Importance.HIGH, HEARTBEAT_INTERVAL_MS_DOC) - .define(CommonClientConfigs.METADATA_MAX_AGE_CONFIG, - ConfigDef.Type.LONG, - 5 * 60 * 1000, - atLeast(0), - ConfigDef.Importance.LOW, - CommonClientConfigs.METADATA_MAX_AGE_DOC) - .define(CommonClientConfigs.CLIENT_ID_CONFIG, - ConfigDef.Type.STRING, - "", - ConfigDef.Importance.LOW, - CommonClientConfigs.CLIENT_ID_DOC) - .define(CommonClientConfigs.SEND_BUFFER_CONFIG, - ConfigDef.Type.INT, - 128 * 1024, - atLeast(0), - ConfigDef.Importance.MEDIUM, - CommonClientConfigs.SEND_BUFFER_DOC) - .define(CommonClientConfigs.RECEIVE_BUFFER_CONFIG, - ConfigDef.Type.INT, - 32 * 1024, - atLeast(0), - ConfigDef.Importance.MEDIUM, - CommonClientConfigs.RECEIVE_BUFFER_DOC) - .define(CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG, - ConfigDef.Type.LONG, - 50L, - atLeast(0L), - ConfigDef.Importance.LOW, - CommonClientConfigs.RECONNECT_BACKOFF_MS_DOC) - .define(CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_CONFIG, - ConfigDef.Type.LONG, - 1000L, - atLeast(0L), - ConfigDef.Importance.LOW, - CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_DOC) - .define(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG, - ConfigDef.Type.LONG, - 100L, - atLeast(0L), - ConfigDef.Importance.LOW, - CommonClientConfigs.RETRY_BACKOFF_MS_DOC) - .define(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG, - ConfigDef.Type.INT, - 40 * 1000, - atLeast(0), - ConfigDef.Importance.MEDIUM, - CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC) - /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */ - .define(CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG, - ConfigDef.Type.LONG, - 9 * 60 * 1000, - ConfigDef.Importance.MEDIUM, - CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC) + + .define(CommonClientConfigDefs.metadataMaxAge(5 * 60 * 1000L)) + .define(CommonClientConfigDefs.clientId(CLIENT_ID_DOC)) + .define(CommonClientConfigDefs.sendBufferBytes(128 * 1024)) + .define(CommonClientConfigDefs.receiveBufferBytes(64 * 1024)) + .define(CommonClientConfigDefs.reconnectBackoffMs(50L)) + .define(CommonClientConfigDefs.reconnectBackoffMaxMs(1000L)) + .define(CommonClientConfigDefs.retryBackoffMs(100L)) + .define(CommonClientConfigDefs.requestTimeoutMs(40 * 1000,CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC)) + /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */ + .define(CommonClientConfigDefs.connectionsMaxIdleMs(9 * 60 * 1000L)) // security support - .define(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, - ConfigDef.Type.STRING, - CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL, - ConfigDef.Importance.MEDIUM, - CommonClientConfigs.SECURITY_PROTOCOL_DOC) + .define(CommonClientConfigDefs.securityProtocol()) + .withClientSslSupport() .withClientSaslSupport() + .define(WORKER_SYNC_TIMEOUT_MS_CONFIG, ConfigDef.Type.INT, 3000, diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java index 0795389593704..0f6d5e242e575 100644 --- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java +++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.streams; +import org.apache.kafka.clients.CommonClientConfigDefs; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.AdminClientConfig; @@ -28,7 +29,6 @@ import org.apache.kafka.common.config.ConfigDef.Importance; import org.apache.kafka.common.config.ConfigDef.Type; import org.apache.kafka.common.config.TopicConfig; -import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.errors.DeserializationExceptionHandler; @@ -51,7 +51,6 @@ import java.util.Set; import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; -import static org.apache.kafka.common.config.ConfigDef.Range.between; import static org.apache.kafka.common.config.ConfigDef.ValidString.in; import static org.apache.kafka.common.requests.IsolationLevel.READ_COMMITTED; @@ -393,11 +392,7 @@ public class StreamsConfig extends AbstractConfig { atLeast(0), Importance.MEDIUM, CACHE_MAX_BYTES_BUFFERING_DOC) - .define(CLIENT_ID_CONFIG, - Type.STRING, - "", - Importance.MEDIUM, - CLIENT_ID_DOC) + .define(CommonClientConfigDefs.clientId(CLIENT_ID_DOC)) .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, Type.CLASS, LogAndFailExceptionHandler.class.getName(), @@ -439,11 +434,7 @@ public class StreamsConfig extends AbstractConfig { in(AT_LEAST_ONCE, EXACTLY_ONCE), Importance.MEDIUM, PROCESSING_GUARANTEE_DOC) - .define(SECURITY_PROTOCOL_CONFIG, - Type.STRING, - CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL, - Importance.MEDIUM, - CommonClientConfigs.SECURITY_PROTOCOL_DOC) + .define(CommonClientConfigDefs.securityProtocol()) // LOW @@ -462,40 +453,12 @@ public class StreamsConfig extends AbstractConfig { DEFAULT_COMMIT_INTERVAL_MS, Importance.LOW, COMMIT_INTERVAL_MS_DOC) - .define(CONNECTIONS_MAX_IDLE_MS_CONFIG, - ConfigDef.Type.LONG, - 9 * 60 * 1000, - ConfigDef.Importance.LOW, - CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC) - .define(METADATA_MAX_AGE_CONFIG, - ConfigDef.Type.LONG, - 5 * 60 * 1000, - atLeast(0), - ConfigDef.Importance.LOW, - CommonClientConfigs.METADATA_MAX_AGE_DOC) - .define(METRICS_NUM_SAMPLES_CONFIG, - Type.INT, - 2, - atLeast(1), - Importance.LOW, - CommonClientConfigs.METRICS_NUM_SAMPLES_DOC) - .define(METRIC_REPORTER_CLASSES_CONFIG, - Type.LIST, - "", - Importance.LOW, - CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC) - .define(METRICS_RECORDING_LEVEL_CONFIG, - Type.STRING, - Sensor.RecordingLevel.INFO.toString(), - in(Sensor.RecordingLevel.INFO.toString(), Sensor.RecordingLevel.DEBUG.toString()), - Importance.LOW, - CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC) - .define(METRICS_SAMPLE_WINDOW_MS_CONFIG, - Type.LONG, - 30000, - atLeast(0), - Importance.LOW, - CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_DOC) + .define(CommonClientConfigDefs.connectionsMaxIdleMs(9 * 60 * 1000L)) + .define(CommonClientConfigDefs.metadataMaxAge(5 * 60 * 1000L)) + .define(CommonClientConfigDefs.metricsNumSamplesConfig(2)) + .define(CommonClientConfigDefs.metricReporterClasses()) + .define(CommonClientConfigDefs.metricsRecordingLevel()) + .define(CommonClientConfigDefs.metricsSampleWindowMs(30_000L)) .define(PARTITION_GROUPER_CLASS_CONFIG, Type.CLASS, DefaultPartitionGrouper.class.getName(), @@ -506,53 +469,23 @@ public class StreamsConfig extends AbstractConfig { 100, Importance.LOW, POLL_MS_DOC) - .define(RECEIVE_BUFFER_CONFIG, - Type.INT, - 32 * 1024, - atLeast(0), - Importance.LOW, - CommonClientConfigs.RECEIVE_BUFFER_DOC) - .define(RECONNECT_BACKOFF_MS_CONFIG, - Type.LONG, - 50L, - atLeast(0L), - Importance.LOW, - CommonClientConfigs.RECONNECT_BACKOFF_MS_DOC) - .define(RECONNECT_BACKOFF_MAX_MS_CONFIG, - Type.LONG, - 1000L, - atLeast(0L), - ConfigDef.Importance.LOW, - CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_DOC) - .define(RETRIES_CONFIG, - Type.INT, - 0, - between(0, Integer.MAX_VALUE), - ConfigDef.Importance.LOW, - CommonClientConfigs.RETRIES_DOC) + .define(CommonClientConfigDefs.receiveBufferBytes(32 * 1024)) + .define(CommonClientConfigDefs.reconnectBackoffMs(50L)) + .define(CommonClientConfigDefs.reconnectBackoffMaxMs(1000L)) + .define(CommonClientConfigDefs.retries(0, CommonClientConfigs.RETRIES_DOC)) .define(RETRY_BACKOFF_MS_CONFIG, Type.LONG, 100L, atLeast(0L), ConfigDef.Importance.LOW, CommonClientConfigs.RETRY_BACKOFF_MS_DOC) - .define(REQUEST_TIMEOUT_MS_CONFIG, - Type.INT, - 40 * 1000, - atLeast(0), - ConfigDef.Importance.LOW, - CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC) + .define(CommonClientConfigDefs.requestTimeoutMs(40 * 1000, CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC)) .define(ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, Type.CLASS, null, Importance.LOW, ROCKSDB_CONFIG_SETTER_CLASS_DOC) - .define(SEND_BUFFER_CONFIG, - Type.INT, - 128 * 1024, - atLeast(0), - Importance.LOW, - CommonClientConfigs.SEND_BUFFER_DOC) + .define(CommonClientConfigDefs.sendBufferBytes(128 * 1024)) .define(STATE_CLEANUP_DELAY_MS_CONFIG, Type.LONG, 10 * 60 * 1000, diff --git a/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java b/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java index c5764b49cca0f..3abc0ea32f946 100644 --- a/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java +++ b/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java @@ -18,6 +18,7 @@ import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.clients.CommonClientConfigDefs; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.MetricName; @@ -90,9 +91,9 @@ public class PushHttpMetricsReporter implements MetricsReporter { .define(METRICS_HOST_CONFIG, ConfigDef.Type.STRING, "", ConfigDef.Importance.LOW, "The hostname to report with each metric; if empty, defaults to the FQDN that can be automatically" + "determined") - .define(CLIENT_ID_CONFIG, ConfigDef.Type.STRING, "", ConfigDef.Importance.LOW, + .define(CommonClientConfigDefs.clientId( "Client ID to identify the application, generally inherited from the " + - "producer/consumer/streams/connect instance"); + "producer/consumer/streams/connect instance")); public PushHttpMetricsReporter() { time = new SystemTime();