kafka WorkerConfig 源码
kafka WorkerConfig 代码
文件路径：/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
/*
* 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.connect.runtime;
import org.apache.kafka.clients.ClientDnsLookup;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.common.KafkaFuture;
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.ConfigException;
import org.apache.kafka.common.config.SslClientAuth;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.storage.SimpleHeaderConverter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.regex.Pattern;
import org.eclipse.jetty.util.StringUtil;
import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
import static org.apache.kafka.common.config.ConfigDef.ValidString.in;
import static org.apache.kafka.connect.runtime.SourceConnectorConfig.TOPIC_CREATION_PREFIX;
/**
* Common base class providing configuration for Kafka Connect workers, whether standalone or distributed.
*/
public class WorkerConfig extends AbstractConfig {
private static final Logger log = LoggerFactory.getLogger(WorkerConfig.class);
private static final Pattern COMMA_WITH_WHITESPACE = Pattern.compile("\\s*,\\s*");
private static final Collection<String> HEADER_ACTIONS = Collections.unmodifiableList(
Arrays.asList("set", "add", "setDate", "addDate")
);
public static final String BOOTSTRAP_SERVERS_CONFIG = "bootstrap.servers";
public static final String BOOTSTRAP_SERVERS_DOC
= "A list of host/port pairs to use for establishing the initial connection to the Kafka "
+ "cluster. The client will make use of all servers irrespective of which servers are "
+ "specified here for bootstrapping—this list only impacts the initial hosts used "
+ "to discover the full set of servers. This list should be in the form "
+ "<code>host1:port1,host2:port2,...</code>. Since these servers are just used for the "
+ "initial connection to discover the full cluster membership (which may change "
+ "dynamically), 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_DEFAULT = "localhost:9092";
public static final String CLIENT_DNS_LOOKUP_CONFIG = CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG;
public static final String CLIENT_DNS_LOOKUP_DOC = CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC;
public static final String KEY_CONVERTER_CLASS_CONFIG = "key.converter";
public static final String KEY_CONVERTER_CLASS_DOC =
"Converter class used to convert between Kafka Connect format and the serialized form that is written to Kafka." +
" This controls the format of the keys in messages written to or read from Kafka, and since this is" +
" independent of connectors it allows any connector to work with any serialization format." +
" Examples of common formats include JSON and Avro.";
public static final String VALUE_CONVERTER_CLASS_CONFIG = "value.converter";
public static final String VALUE_CONVERTER_CLASS_DOC =
"Converter class used to convert between Kafka Connect format and the serialized form that is written to Kafka." +
" This controls the format of the values in messages written to or read from Kafka, and since this is" +
" independent of connectors it allows any connector to work with any serialization format." +
" Examples of common formats include JSON and Avro.";
public static final String HEADER_CONVERTER_CLASS_CONFIG = "header.converter";
public static final String HEADER_CONVERTER_CLASS_DOC =
"HeaderConverter class used to convert between Kafka Connect format and the serialized form that is written to Kafka." +
" This controls the format of the header values in messages written to or read from Kafka, and since this is" +
" independent of connectors it allows any connector to work with any serialization format." +
" Examples of common formats include JSON and Avro. By default, the SimpleHeaderConverter is used to serialize" +
" header values to strings and deserialize them by inferring the schemas.";
public static final String HEADER_CONVERTER_CLASS_DEFAULT = SimpleHeaderConverter.class.getName();
public static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_CONFIG
= "task.shutdown.graceful.timeout.ms";
private static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DOC =
"Amount of time to wait for tasks to shutdown gracefully. This is the total amount of time,"
+ " not per task. All task have shutdown triggered, then they are waited on sequentially.";
private static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DEFAULT = "5000";
public static final String OFFSET_COMMIT_INTERVAL_MS_CONFIG = "offset.flush.interval.ms";
private static final String OFFSET_COMMIT_INTERVAL_MS_DOC
= "Interval at which to try committing offsets for tasks.";
public static final long OFFSET_COMMIT_INTERVAL_MS_DEFAULT = 60000L;
public static final String OFFSET_COMMIT_TIMEOUT_MS_CONFIG = "offset.flush.timeout.ms";
private static final String OFFSET_COMMIT_TIMEOUT_MS_DOC
= "Maximum number of milliseconds to wait for records to flush and partition offset data to be"
+ " committed to offset storage before cancelling the process and restoring the offset "
+ "data to be committed in a future attempt. This property has no effect for source connectors "
+ "running with exactly-once support.";
public static final long OFFSET_COMMIT_TIMEOUT_MS_DEFAULT = 5000L;
public static final String LISTENERS_CONFIG = "listeners";
private static final String LISTENERS_DOC
= "List of comma-separated URIs the REST API will listen on. The supported protocols are HTTP and HTTPS.\n" +
" Specify hostname as 0.0.0.0 to bind to all interfaces.\n" +
" Leave hostname empty to bind to default interface.\n" +
" Examples of legal listener lists: HTTP://myhost:8083,HTTPS://myhost:8084";
static final List<String> LISTENERS_DEFAULT = Collections.singletonList("http://:8083");
public static final String REST_ADVERTISED_HOST_NAME_CONFIG = "rest.advertised.host.name";
private static final String REST_ADVERTISED_HOST_NAME_DOC
= "If this is set, this is the hostname that will be given out to other workers to connect to.";
public static final String REST_ADVERTISED_PORT_CONFIG = "rest.advertised.port";
private static final String REST_ADVERTISED_PORT_DOC
= "If this is set, this is the port that will be given out to other workers to connect to.";
public static final String REST_ADVERTISED_LISTENER_CONFIG = "rest.advertised.listener";
private static final String REST_ADVERTISED_LISTENER_DOC
= "Sets the advertised listener (HTTP or HTTPS) which will be given to other workers to use.";
public static final String ACCESS_CONTROL_ALLOW_ORIGIN_CONFIG = "access.control.allow.origin";
protected static final String ACCESS_CONTROL_ALLOW_ORIGIN_DOC =
"Value to set the Access-Control-Allow-Origin header to for REST API requests." +
"To enable cross origin access, set this to the domain of the application that should be permitted" +
" to access the API, or '*' to allow access from any domain. The default value only allows access" +
" from the domain of the REST API.";
protected static final String ACCESS_CONTROL_ALLOW_ORIGIN_DEFAULT = "";
public static final String ACCESS_CONTROL_ALLOW_METHODS_CONFIG = "access.control.allow.methods";
protected static final String ACCESS_CONTROL_ALLOW_METHODS_DOC =
"Sets the methods supported for cross origin requests by setting the Access-Control-Allow-Methods header. "
+ "The default value of the Access-Control-Allow-Methods header allows cross origin requests for GET, POST and HEAD.";
protected static final String ACCESS_CONTROL_ALLOW_METHODS_DEFAULT = "";
public static final String ADMIN_LISTENERS_CONFIG = "admin.listeners";
protected static final String ADMIN_LISTENERS_DOC = "List of comma-separated URIs the Admin REST API will listen on." +
" The supported protocols are HTTP and HTTPS." +
" An empty or blank string will disable this feature." +
" The default behavior is to use the regular listener (specified by the 'listeners' property).";
public static final String ADMIN_LISTENERS_HTTPS_CONFIGS_PREFIX = "admin.listeners.https.";
public static final String PLUGIN_PATH_CONFIG = "plugin.path";
protected static final String PLUGIN_PATH_DOC = "List of paths separated by commas (,) that "
+ "contain plugins (connectors, converters, transformations). The list should consist"
+ " of top level directories that include any combination of: \n"
+ "a) directories immediately containing jars with plugins and their dependencies\n"
+ "b) uber-jars with plugins and their dependencies\n"
+ "c) directories immediately containing the package directory structure of classes of "
+ "plugins and their dependencies\n"
+ "Note: symlinks will be followed to discover dependencies or plugins.\n"
+ "Examples: plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,"
+ "/opt/connectors\n"
+ "Do not use config provider variables in this property, since the raw path is used "
+ "by the worker's scanner before config providers are initialized and used to "
+ "replace variables.";
public static final String CONFIG_PROVIDERS_CONFIG = "config.providers";
protected static final String CONFIG_PROVIDERS_DOC =
"Comma-separated names of <code>ConfigProvider</code> classes, loaded and used "
+ "in the order specified. Implementing the interface "
+ "<code>ConfigProvider</code> allows you to replace variable references in connector configurations, "
+ "such as for externalized secrets. ";
public static final String REST_EXTENSION_CLASSES_CONFIG = "rest.extension.classes";
protected static final String REST_EXTENSION_CLASSES_DOC =
"Comma-separated names of <code>ConnectRestExtension</code> classes, loaded and called "
+ "in the order specified. Implementing the interface "
+ "<code>ConnectRestExtension</code> allows you to inject into Connect's REST API user defined resources like filters. "
+ "Typically used to add custom capability like logging, security, etc. ";
public static final String CONNECTOR_CLIENT_POLICY_CLASS_CONFIG = "connector.client.config.override.policy";
public static final String CONNECTOR_CLIENT_POLICY_CLASS_DOC =
"Class name or alias of implementation of <code>ConnectorClientConfigOverridePolicy</code>. Defines what client configurations can be "
+ "overridden by the connector. The default implementation is `All`, meaning connector configurations can override all client properties. "
+ "The other possible policies in the framework include `None` to disallow connectors from overriding client properties, "
+ "and `Principal` to allow connectors to override only client principals.";
public static final String CONNECTOR_CLIENT_POLICY_CLASS_DEFAULT = "All";
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;
@Deprecated
public static final String AUTO_INCLUDE_JMX_REPORTER_CONFIG = CommonClientConfigs.AUTO_INCLUDE_JMX_REPORTER_CONFIG;
public static final String TOPIC_TRACKING_ENABLE_CONFIG = "topic.tracking.enable";
protected static final String TOPIC_TRACKING_ENABLE_DOC = "Enable tracking the set of active "
+ "topics per connector during runtime.";
protected static final boolean TOPIC_TRACKING_ENABLE_DEFAULT = true;
public static final String TOPIC_TRACKING_ALLOW_RESET_CONFIG = "topic.tracking.allow.reset";
protected static final String TOPIC_TRACKING_ALLOW_RESET_DOC = "If set to true, it allows "
+ "user requests to reset the set of active topics per connector.";
protected static final boolean TOPIC_TRACKING_ALLOW_RESET_DEFAULT = true;
public static final String CONNECT_KAFKA_CLUSTER_ID = "connect.kafka.cluster.id";
public static final String CONNECT_GROUP_ID = "connect.group.id";
public static final String TOPIC_CREATION_ENABLE_CONFIG = "topic.creation.enable";
protected static final String TOPIC_CREATION_ENABLE_DOC = "Whether to allow "
+ "automatic creation of topics used by source connectors, when source connectors "
+ "are configured with `" + TOPIC_CREATION_PREFIX + "` properties. Each task will use an "
+ "admin client to create its topics and will not depend on the Kafka brokers "
+ "to create topics automatically.";
protected static final boolean TOPIC_CREATION_ENABLE_DEFAULT = true;
public static final String RESPONSE_HTTP_HEADERS_CONFIG = "response.http.headers.config";
protected static final String RESPONSE_HTTP_HEADERS_DOC = "Rules for REST API HTTP response headers";
protected static final String RESPONSE_HTTP_HEADERS_DEFAULT = "";
/**
* Get a basic ConfigDef for a WorkerConfig. This includes all the common settings. Subclasses can use this to
* bootstrap their own ConfigDef.
* @return a ConfigDef with all the common options specified
*/
protected static ConfigDef baseConfigDef() {
return new ConfigDef()
.define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, BOOTSTRAP_SERVERS_DEFAULT,
Importance.HIGH, BOOTSTRAP_SERVERS_DOC)
.define(CLIENT_DNS_LOOKUP_CONFIG,
Type.STRING,
ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
in(ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
Importance.MEDIUM,
CLIENT_DNS_LOOKUP_DOC)
.define(KEY_CONVERTER_CLASS_CONFIG, Type.CLASS,
Importance.HIGH, KEY_CONVERTER_CLASS_DOC)
.define(VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS,
Importance.HIGH, VALUE_CONVERTER_CLASS_DOC)
.define(TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_CONFIG, Type.LONG,
TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DEFAULT, Importance.LOW,
TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DOC)
.define(OFFSET_COMMIT_INTERVAL_MS_CONFIG, Type.LONG, OFFSET_COMMIT_INTERVAL_MS_DEFAULT,
Importance.LOW, OFFSET_COMMIT_INTERVAL_MS_DOC)
.define(OFFSET_COMMIT_TIMEOUT_MS_CONFIG, Type.LONG, OFFSET_COMMIT_TIMEOUT_MS_DEFAULT,
Importance.LOW, OFFSET_COMMIT_TIMEOUT_MS_DOC)
.define(LISTENERS_CONFIG, Type.LIST, LISTENERS_DEFAULT, new ListenersValidator(), Importance.LOW, LISTENERS_DOC)
.define(REST_ADVERTISED_HOST_NAME_CONFIG, Type.STRING, null, Importance.LOW, REST_ADVERTISED_HOST_NAME_DOC)
.define(REST_ADVERTISED_PORT_CONFIG, Type.INT, null, Importance.LOW, REST_ADVERTISED_PORT_DOC)
.define(REST_ADVERTISED_LISTENER_CONFIG, Type.STRING, null, Importance.LOW, REST_ADVERTISED_LISTENER_DOC)
.define(ACCESS_CONTROL_ALLOW_ORIGIN_CONFIG, Type.STRING,
ACCESS_CONTROL_ALLOW_ORIGIN_DEFAULT, Importance.LOW,
ACCESS_CONTROL_ALLOW_ORIGIN_DOC)
.define(ACCESS_CONTROL_ALLOW_METHODS_CONFIG, Type.STRING,
ACCESS_CONTROL_ALLOW_METHODS_DEFAULT, Importance.LOW,
ACCESS_CONTROL_ALLOW_METHODS_DOC)
.define(PLUGIN_PATH_CONFIG,
Type.LIST,
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(AUTO_INCLUDE_JMX_REPORTER_CONFIG,
Type.BOOLEAN,
true,
Importance.LOW,
CommonClientConfigs.AUTO_INCLUDE_JMX_REPORTER_DOC)
.define(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG,
ConfigDef.Type.STRING, SslClientAuth.NONE.toString(), in(Utils.enumOptions(SslClientAuth.class)), ConfigDef.Importance.LOW, BrokerSecurityConfigs.SSL_CLIENT_AUTH_DOC)
.define(HEADER_CONVERTER_CLASS_CONFIG, Type.CLASS,
HEADER_CONVERTER_CLASS_DEFAULT,
Importance.LOW, HEADER_CONVERTER_CLASS_DOC)
.define(CONFIG_PROVIDERS_CONFIG, Type.LIST,
Collections.emptyList(),
Importance.LOW, CONFIG_PROVIDERS_DOC)
.define(REST_EXTENSION_CLASSES_CONFIG, Type.LIST, "",
Importance.LOW, REST_EXTENSION_CLASSES_DOC)
.define(ADMIN_LISTENERS_CONFIG, Type.LIST, null,
new AdminListenersValidator(), Importance.LOW, ADMIN_LISTENERS_DOC)
.define(CONNECTOR_CLIENT_POLICY_CLASS_CONFIG, Type.STRING, CONNECTOR_CLIENT_POLICY_CLASS_DEFAULT,
Importance.MEDIUM, CONNECTOR_CLIENT_POLICY_CLASS_DOC)
.define(TOPIC_TRACKING_ENABLE_CONFIG, Type.BOOLEAN, TOPIC_TRACKING_ENABLE_DEFAULT,
Importance.LOW, TOPIC_TRACKING_ENABLE_DOC)
.define(TOPIC_TRACKING_ALLOW_RESET_CONFIG, Type.BOOLEAN, TOPIC_TRACKING_ALLOW_RESET_DEFAULT,
Importance.LOW, TOPIC_TRACKING_ALLOW_RESET_DOC)
.define(TOPIC_CREATION_ENABLE_CONFIG, Type.BOOLEAN, TOPIC_CREATION_ENABLE_DEFAULT, Importance.LOW,
TOPIC_CREATION_ENABLE_DOC)
.define(RESPONSE_HTTP_HEADERS_CONFIG, Type.STRING, RESPONSE_HTTP_HEADERS_DEFAULT,
new ResponseHttpHeadersValidator(), Importance.LOW, RESPONSE_HTTP_HEADERS_DOC)
// security support
.withClientSslSupport();
}
private String kafkaClusterId;
public static String lookupKafkaClusterId(WorkerConfig config) {
log.info("Creating Kafka admin client");
try (Admin adminClient = Admin.create(config.originals())) {
return lookupKafkaClusterId(adminClient);
}
}
static String lookupKafkaClusterId(Admin adminClient) {
log.debug("Looking up Kafka cluster ID");
try {
KafkaFuture<String> clusterIdFuture = adminClient.describeCluster().clusterId();
if (clusterIdFuture == null) {
log.info("Kafka cluster version is too old to return cluster ID");
return null;
}
log.debug("Fetching Kafka cluster ID");
String kafkaClusterId = clusterIdFuture.get();
log.info("Kafka cluster ID: {}", kafkaClusterId);
return kafkaClusterId;
} catch (InterruptedException e) {
throw new ConnectException("Unexpectedly interrupted when looking up Kafka cluster info", e);
} catch (ExecutionException e) {
throw new ConnectException("Failed to connect to and describe Kafka cluster. "
+ "Check worker's broker connection and security properties.", e);
}
}
private void logInternalConverterRemovalWarnings(Map<String, String> props) {
List<String> removedProperties = new ArrayList<>();
for (String property : Arrays.asList("internal.key.converter", "internal.value.converter")) {
if (props.containsKey(property)) {
removedProperties.add(property);
}
removedProperties.addAll(originalsWithPrefix(property + ".").keySet());
}
if (!removedProperties.isEmpty()) {
log.warn(
"The worker has been configured with one or more internal converter properties ({}). "
+ "Support for these properties was deprecated in version 2.0 and removed in version 3.0, "
+ "and specifying them will have no effect. "
+ "Instead, an instance of the JsonConverter with schemas.enable "
+ "set to false will be used. For more information, please visit "
+ "https://kafka.apache.org/documentation/#upgrade and consult the upgrade notes"
+ "for the 3.0 release.",
removedProperties);
}
}
private void logPluginPathConfigProviderWarning(Map<String, String> rawOriginals) {
String rawPluginPath = rawOriginals.get(PLUGIN_PATH_CONFIG);
// Can't use AbstractConfig::originalsStrings here since some values may be null, which
// causes that method to fail
String transformedPluginPath = Objects.toString(originals().get(PLUGIN_PATH_CONFIG));
if (!Objects.equals(rawPluginPath, transformedPluginPath)) {
log.warn(
"Variables cannot be used in the 'plugin.path' property, since the property is "
+ "used by plugin scanning before the config providers that replace the "
+ "variables are initialized. The raw value '{}' was used for plugin scanning, as "
+ "opposed to the transformed value '{}', and this may cause unexpected results.",
rawPluginPath,
transformedPluginPath
);
}
}
/**
* @return the {@link CommonClientConfigs#BOOTSTRAP_SERVERS_CONFIG bootstrap servers} property
* used by the worker when instantiating Kafka clients for connectors and tasks (unless overridden)
* and its internal topics (if running in distributed mode)
*/
public String bootstrapServers() {
return String.join(",", getList(BOOTSTRAP_SERVERS_CONFIG));
}
public Integer getRebalanceTimeout() {
return null;
}
public boolean topicCreationEnable() {
return getBoolean(TOPIC_CREATION_ENABLE_CONFIG);
}
/**
* Whether this worker is configured with exactly-once support for source connectors.
* The default implementation returns {@code false} and should be overridden by subclasses
* if the worker mode for the subclass provides exactly-once support for source connectors.
* @return whether exactly-once support is enabled for source connectors on this worker
*/
public boolean exactlyOnceSourceEnabled() {
return false;
}
/**
* Get the internal topic used by this worker to store source connector offsets.
* The default implementation returns {@code null} and should be overridden by subclasses
* if the worker mode for the subclass uses an internal offsets topic.
* @return the name of the internal offsets topic, or {@code null} if the worker does not use
* an internal offsets topic
*/
public String offsetsTopic() {
return null;
}
/**
* Determine whether this worker supports per-connector source offsets topics.
* The default implementation returns {@code false} and should be overridden by subclasses
* if the worker mode for the subclass supports per-connector offsets topics.
* @return whether the worker supports per-connector offsets topics
*/
public boolean connectorOffsetsTopicsPermitted() {
return false;
}
/**
* @return the offset commit interval for tasks created by this worker
*/
public long offsetCommitInterval() {
return getLong(OFFSET_COMMIT_INTERVAL_MS_CONFIG);
}
/**
* Get the {@link CommonClientConfigs#GROUP_ID_CONFIG group ID} used by this worker to form a cluster.
* The default implementation returns {@code null} and should be overridden by subclasses
* if the worker mode for the subclass is capable of forming a cluster using Kafka's group coordination API.
* @return the group ID for the worker's cluster, or {@code null} if the worker is not capable of forming a cluster.
*/
public String groupId() {
return null;
}
public String kafkaClusterId() {
if (kafkaClusterId == null) {
kafkaClusterId = lookupKafkaClusterId(this);
}
return kafkaClusterId;
}
@Override
protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
return CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
}
public static List<String> pluginLocations(Map<String, String> props) {
String locationList = props.get(WorkerConfig.PLUGIN_PATH_CONFIG);
return locationList == null
? new ArrayList<>()
: Arrays.asList(COMMA_WITH_WHITESPACE.split(locationList.trim(), -1));
}
public WorkerConfig(ConfigDef definition, Map<String, String> props) {
super(definition, props);
logInternalConverterRemovalWarnings(props);
logPluginPathConfigProviderWarning(props);
}
// Visible for testing
static void validateHttpResponseHeaderConfig(String config) {
try {
// validate format
String[] configTokens = config.trim().split("\\s+", 2);
if (configTokens.length != 2) {
throw new ConfigException(String.format("Invalid format of header config '%s'. "
+ "Expected: '[action] [header name]:[header value]'", config));
}
// validate action
String method = configTokens[0].trim();
validateHeaderConfigAction(method);
// validate header name and header value pair
String header = configTokens[1];
String[] headerTokens = header.trim().split(":");
if (headerTokens.length != 2) {
throw new ConfigException(
String.format("Invalid format of header name and header value pair '%s'. "
+ "Expected: '[header name]:[header value]'", header));
}
// validate header name
String headerName = headerTokens[0].trim();
if (headerName.isEmpty() || headerName.matches(".*\\s+.*")) {
throw new ConfigException(String.format("Invalid header name '%s'. "
+ "The '[header name]' cannot contain whitespace", headerName));
}
} catch (ArrayIndexOutOfBoundsException e) {
throw new ConfigException(String.format("Invalid header config '%s'.", config), e);
}
}
// Visible for testing
static void validateHeaderConfigAction(String action) {
if (HEADER_ACTIONS.stream().noneMatch(action::equalsIgnoreCase)) {
throw new ConfigException(String.format("Invalid header config action: '%s'. "
+ "Expected one of %s", action, HEADER_ACTIONS));
}
}
private static class ListenersValidator implements ConfigDef.Validator {
@Override
public void ensureValid(String name, Object value) {
if (!(value instanceof List)) {
throw new ConfigException("Invalid value type for listeners (expected list of URLs , ex: http://localhost:8080,https://localhost:8443).");
}
List<?> items = (List<?>) value;
if (items.isEmpty()) {
throw new ConfigException("Invalid value for listeners, at least one URL is expected, ex: http://localhost:8080,https://localhost:8443.");
}
for (Object item : items) {
if (!(item instanceof String)) {
throw new ConfigException("Invalid type for listeners (expected String).");
}
if (Utils.isBlank((String) item)) {
throw new ConfigException("Empty URL found when parsing listeners list.");
}
}
}
@Override
public String toString() {
return "List of comma-separated URLs, ex: http://localhost:8080,https://localhost:8443.";
}
}
private static class AdminListenersValidator implements ConfigDef.Validator {
@Override
public void ensureValid(String name, Object value) {
if (value == null) {
return;
}
if (!(value instanceof List)) {
throw new ConfigException("Invalid value type for admin.listeners (expected list).");
}
List<?> items = (List<?>) value;
if (items.isEmpty()) {
return;
}
for (Object item : items) {
if (!(item instanceof String)) {
throw new ConfigException("Invalid type for admin.listeners (expected String).");
}
if (Utils.isBlank((String) item)) {
throw new ConfigException("Empty URL found when parsing admin.listeners list.");
}
}
}
@Override
public String toString() {
return "List of comma-separated URLs, ex: http://localhost:8080,https://localhost:8443.";
}
}
private static class ResponseHttpHeadersValidator implements ConfigDef.Validator {
@Override
public void ensureValid(String name, Object value) {
String strValue = (String) value;
if (Utils.isBlank(strValue)) {
return;
}
String[] configs = StringUtil.csvSplit(strValue); // handles and removed surrounding quotes
Arrays.stream(configs).forEach(WorkerConfig::validateHttpResponseHeaderConfig);
}
@Override
public String toString() {
return "Comma-separated header rules, where each header rule is of the form "
+ "'[action] [header name]:[header value]' and optionally surrounded by double quotes "
+ "if any part of a header rule contains a comma";
}
}
}
相关信息
相关文章
kafka AbstractWorkerSourceTask 源码
kafka CloseableConnectorContext 源码
0
赞
热门推荐
-
2、 - 优质文章
-
3、 gate.io
-
7、 golang
-
9、 openharmony
-
10、 Vue中input框自动聚焦