Skip to content

KAFKA-6891 fixed config constraints in KafkaConnect + refactoring #4995

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
oleg-smith wants to merge 1 commit into from
Closed

KAFKA-6891 fixed config constraints in KafkaConnect + refactoring #4995

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
187 changes: 187 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/CommonClientConfigDefs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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);
}
}
100 changes: 19 additions & 81 deletions clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*/
Expand All @@ -40,19 +35,16 @@ public class AdminClientConfig extends AbstractConfig {
* <code>bootstrap.servers</code>
*/
public static final String BOOTSTRAP_SERVERS_CONFIG = CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
private static final String BOOTSTRAP_SERVERS_DOC = CommonClientConfigs.BOOTSTRAP_SERVERS_DOC;

/**
* <code>reconnect.backoff.ms</code>
*/
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;

/**
* <code>reconnect.backoff.max.ms</code>
*/
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;

/**
* <code>retry.backoff.ms</code>
Expand All @@ -64,106 +56,52 @@ public class AdminClientConfig extends AbstractConfig {

/** <code>connections.max.idle.ms</code> */
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;

/** <code>request.timeout.ms</code> */
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();
}
Expand Down
Loading