kafka内存溢出分析：OutOfMemoryError
·
线上Kafka问题分析：OutOfMemoryError
前言：
最近在线上遇到个奇怪的问题，具体的情景是：由于客户有相应的需求，我们需要提前在程序中判断Kafka集群是否可用，于是我们使用了Kafka0.11.0.0版本之后提供的KafkaAdminClient，这可以方便的创建一个kafka客户端进行集群的管理操作。
经过：
我们使用listTopics()来获取kafka集群所有的topic，如果能获取到相应数据，则能够说明kafka集群运行正常，代码很简单：
package com.wenjie.demo.kafkaconnector.client;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
/**
* @author BlackSugar
* @date 2020/11/19
*/
@Slf4j
public class MyAdminClient {
private static InetAddress localInetAddr = null;
static {
try {
localInetAddr = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
e.printStackTrace();
log.error("Get local host error!!!");
}
}
private static void isReachable(InetAddress remoteInetAddr, int port, int timeout) {
Properties properties = new Properties();
String host = remoteInetAddr.getHostAddress() + ":" + port;
properties.put("bootstrap.servers", host);
properties.put("connections.max.idle.ms", 10000);
properties.put("request.timeout.ms", timeout);
try (AdminClient client = KafkaAdminClient.create(properties)) {
ListTopicsResult topics = client.listTopics();
Set<String> names = topics.names().get();
log.info("SUCCESS - Connection established! Local:{}, remote:{}, port:{}, topic:[{}]",
localInetAddr.getHostAddress(), remoteInetAddr.getHostAddress(), port, names);
} catch (InterruptedException | ExecutionException e) {
// Kafka is not available
log.error("FAILED - Can not connect to kafka cluster! Local:{}, remote:{}, port:{}, exception: ",
localInetAddr.getHostAddress(), remoteInetAddr.getHostAddress(), port, e);
}
}
public static void main(String[] args) {
try {
isReachable(InetAddress.getByName("127.0.0.1"), 9092, 5000);
} catch (UnknownHostException e) {
e.printStackTrace();
}
}
}
仅依赖kafka客户端即可，POM文件：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.3.1</version>
</dependency>
<!--lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
成功日志（返回topics）：
2020-11-18 16:33:25.085 INFO 45508 --- [http-nio-8990-exec-8] o.a.kafka.common.utils.AppInfoParser : Kafka startTimeMs: 1605774805085
2020-11-18 16:33:25.096 INFO 45508 --- [http-nio-8990-exec-8] com.wenjie.demo.kafkaconnector.client.MyAdminClient : SUCCESS - Connection established! Local:127.0.0.1, remote:127.0.0.1, port:9092, topic:[topic1,topic2]
失败日志(超时)：
15:47:15.942 [main] DEBUG org.apache.kafka.clients.admin.KafkaAdminClient - [AdminClient clientId=adminclient-1] Kafka admin client closed.
15:47:15.942 [main] ERROR com.wenjie.demo.kafkaconnector.client.MyAdminClient - FAILED - Can not connect to kafka cluster! Local:127.0.0.1, remote:127.0.0.1, port:9091, exception:
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
at com.wenjie.demo.kafkaconnector.client.MyAdminClient.isReachable(MyAdminClient.java:39)
at com.wenjie.demo.kafkaconnector.client.MyAdminClient.main(MyAdminClient.java:51)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.
项目运行了一段时间没什么问题，直到有一天线上突然报出OOM:
2020-11-18 17:37:48.107 ERROR 41072 --- [kafka-admin-client-thread | adminclient-5] o.apache.kafka.common.utils.KafkaThread : Uncaught exception in thread 'kafka-admin-client-thread | adminclient-5':
java.lang.OutOfMemoryError: Java heap space
at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
at org.apache.kafka.common.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30)
at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:112)
at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:424)
at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:385)
at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:651)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:572)
at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:539)
at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1152)
at java.lang.Thread.run(Thread.java:745)
通过观察gc日志，我们发现异常出现的时间进行了Full GC，还GC不掉：
2020-11-18T17:37:48.387+0800: 3.095: [Full GC (Metadata GC Threshold) [PSYoungGen: 6468K->0K(299008K)] [ParOldGen: 5580K->11448K(349696K)] 12049K->11448K(648704K), [Metaspace: 33895K->33895K(1079296K)], 0.0712752 secs] [Times: user=0.20 sys=0.00, real=0.08 secs]
这就奇怪了，怎么会突然报Full GC呢，看GC应该是出现了大对象，直接把内存撑爆了，于是我开始向这个方向着手排查。
通过一系列的验证，我发现出现OOM的时候，都是Kafka客户端尝试访问一个非Kafka的TCP服务的时候（访问正常的Kafka服务会success，访问不畅通的TCP服务的时候会timout）。
顺着这个情况找下去，最终找到了这个类：NetworkReceive，关键代码如下：
public long readFrom(ScatteringByteChannel channel) throws IOException {
...
if (!size.hasRemaining()) {
size.rewind();
int receiveSize = size.getInt();
if (receiveSize < 0)
throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");
if (maxSize != UNLIMITED && receiveSize > maxSize)
throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");
requestedBufferSize = receiveSize;
if (receiveSize == 0) {
buffer = EMPTY_BUFFER;
}
}
...
if (buffer == null && requestedBufferSize != -1) {
buffer = memoryPool.tryAllocate(requestedBufferSize);
...
}
关键就在于这个buffer，在OOM时想要申请一个1213486160字节（约1.1G）的内存空间，而我们程序的最大堆-Xmx才配的1G，当然会OOM啦。
好的，现在问题是找到了 ，那么原因呢？
拜托了万能的google，我找到了这个：
其中表示当Kafka服务端接受到一个HTTP服务的时候，生成的数据会特别大，然后Kafka在0.10.1.1之后做了更新，解决了这个问题，可是客户端呢？就被遗忘了？坑爹呢！
不过可能Kafka的开发人员也没想到会有这么奇葩的需求吧~
这里我推测是由于无法正常序列化HTTP报文，导致序列化出来的数据特别大（ps：thrift也有类似问题）。
言归正传，原因也发现了，那怎么解决？
回到上面代码，我们可以看到：
if (maxSize != UNLIMITED && receiveSize > maxSize)
throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");
requestedBufferSize = receiveSize;
好家伙，我直接好家伙，buffer是根据requestedBufferSize来申请的，这不是有相关配置吗，如果这里我们配置一个maxSize，那么程序就会直接抛出异常，而不会到申请内存的地方，我们再在外部将这个异常捕获处理，那一切不就ok了吗？
naive，我是真的naive，我一路追寻这个maxSize，最终找到了它配置的地方： org.apache.kafka.common.network.Selector，它的构造函数：
public Selector(int maxReceiveSize,
long connectionMaxIdleMs,
int failedAuthenticationDelayMs,
Metrics metrics,
Time time,
String metricGrpPrefix,
Map<String, String> metricTags,
boolean metricsPerConnection,
boolean recordTimePerConnection,
ChannelBuilder channelBuilder,
MemoryPool memoryPool,
LogContext logContext) {
try {
this.nioSelector = java.nio.channels.Selector.open();
} catch (IOException e) {
throw new KafkaException(e);
}
this.maxReceiveSize = maxReceiveSize;
this.time = time;
this.channels = new HashMap<>();
this.explicitlyMutedChannels = new HashSet<>();
this.outOfMemory = false;
this.completedSends = new ArrayList<>();
this.completedReceives = new ArrayList<>();
this.stagedReceives = new HashMap<>();
this.immediatelyConnectedKeys = new HashSet<>();
this.closingChannels = new HashMap<>();
this.keysWithBufferedRead = new HashSet<>();
this.connected = new ArrayList<>();
this.disconnected = new HashMap<>();
this.failedSends = new ArrayList<>();
this.sensors = new SelectorMetrics(metrics, metricGrpPrefix, metricTags, metricsPerConnection);
this.channelBuilder = channelBuilder;
this.recordTimePerConnection = recordTimePerConnection;
this.idleExpiryManager = connectionMaxIdleMs < 0 ? null : new IdleExpiryManager(time, connectionMaxIdleMs);
this.memoryPool = memoryPool;
this.lowMemThreshold = (long) (0.1 * this.memoryPool.size());
this.log = logContext.logger(Selector.class);
this.failedAuthenticationDelayMs = failedAuthenticationDelayMs;
this.delayedClosingChannels = (failedAuthenticationDelayMs > NO_FAILED_AUTHENTICATION_DELAY) ? new LinkedHashMap<String, DelayedAuthenticationFailureClose>() : null;
}
public Selector(int maxReceiveSize,
long connectionMaxIdleMs,
Metrics metrics,
Time time,
String metricGrpPrefix,
Map<String, String> metricTags,
boolean metricsPerConnection,
boolean recordTimePerConnection,
ChannelBuilder channelBuilder,
MemoryPool memoryPool,
LogContext logContext) {
this(maxReceiveSize, connectionMaxIdleMs, NO_FAILED_AUTHENTICATION_DELAY, metrics, time, metricGrpPrefix, metricTags,
metricsPerConnection, recordTimePerConnection, channelBuilder, memoryPool, logContext);
}
public Selector(int maxReceiveSize,
long connectionMaxIdleMs,
int failedAuthenticationDelayMs,
Metrics metrics,
Time time,
String metricGrpPrefix,
Map<String, String> metricTags,
boolean metricsPerConnection,
ChannelBuilder channelBuilder,
LogContext logContext) {
this(maxReceiveSize, connectionMaxIdleMs, failedAuthenticationDelayMs, metrics, time, metricGrpPrefix, metricTags, metricsPerConnection, false, channelBuilder, MemoryPool.NONE, logContext);
}
public Selector(int maxReceiveSize,
long connectionMaxIdleMs,
Metrics metrics,
Time time,
String metricGrpPrefix,
Map<String, String> metricTags,
boolean metricsPerConnection,
ChannelBuilder channelBuilder,
LogContext logContext) {
this(maxReceiveSize, connectionMaxIdleMs, NO_FAILED_AUTHENTICATION_DELAY, metrics, time, metricGrpPrefix, metricTags, metricsPerConnection, channelBuilder, logContext);
}
public Selector(long connectionMaxIdleMS, Metrics metrics, Time time, String metricGrpPrefix, ChannelBuilder channelBuilder, LogContext logContext) {
this(NetworkReceive.UNLIMITED, connectionMaxIdleMS, metrics, time, metricGrpPrefix, Collections.emptyMap(), true, channelBuilder, logContext);
}
}
这个maxReceiveSize就是我们找的目标了，而有他的构造函数一个调用都没有，而没有这个参数的构造函数，它的调用方是：
好吧，一家人整整齐齐。也是不知道为什么Kafka官方不提供一个可配置maxReceiveSize的调用。
我们来看KafkaAdminClient：
static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcessorFactory timeoutProcessorFactory) {
...
Selector(config.getLong(AdminClientConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
metrics, time, metricGrpPrefix, channelBuilder, logContext);
...
}
的确没法配置，没办法，我们只有重写这个类了，这里我们只需要create、listTopics、close方法，其他的就不需要写了，最终代码如下（注意引用我们自己的类）：
最终代码：
package com.wenjie.demo.kafkaconnector.client;
import com.wenjie.demo.kafkaconnector.component.AdminClient;
import com.wenjie.demo.kafkaconnector.component.KafkaAdminClient;
import com.wenjie.demo.kafkaconnector.component.ListTopicsResult;
import lombok.extern.slf4j.Slf4j;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;
import java.util.Set;
/**
* @author BlackSugar
* @date 2020/11/19
*/
@Slf4j
public class MyAdminClient {
private static InetAddress localInetAddr = null;
static {
try {
localInetAddr = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
e.printStackTrace();
log.error("Get local host error!!!");
}
}
private static void isReachable(InetAddress remoteInetAddr, int port, int timeout) {
Properties properties = new Properties();
String host = remoteInetAddr.getHostAddress() + ":" + port;
properties.put("bootstrap.servers", host);
properties.put("connections.max.idle.ms", 10000);
properties.put("request.timeout.ms", timeout);
//重连间隔
properties.put("reconnect.backoff.ms", 1000);
//我们自定义的配置，大小为1m
properties.put("max.buffer.bytes", 1 << 20);
try (AdminClient client = KafkaAdminClient.create(properties)) {
ListTopicsResult topics = client.listTopics();
Set<String> names = topics.names().get();
log.info("SUCCESS - Connection established! Local:{}, remote:{}, port:{}, topic:{}",
localInetAddr.getHostAddress(), remoteInetAddr.getHostAddress(), port, names);
} catch (Exception e) {
// Kafka is not available
log.error("FAILED - Can not connect to kafka cluster! Local:{}, remote:{}, port:{}, exception: ",
localInetAddr.getHostAddress(), remoteInetAddr.getHostAddress(), port, e);
}
}
public static void main(String[] args) {
try {
isReachable(InetAddress.getByName("127.0.0.1"), 9091, 3000);
} catch (UnknownHostException e) {
e.printStackTrace();
}
}
}
代码结构：
其他类：
package com.wenjie.demo.kafkaconnector.component;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.common.annotation.InterfaceStability;
import java.time.Duration;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
@InterfaceStability.Evolving
public abstract class AdminClient implements AutoCloseable {
public static AdminClient create(Properties props) {
return KafkaAdminClient.createInternal(new AdminClientConfig(props, true), null);
}
public static AdminClient create(Map<String, Object> conf) {
return KafkaAdminClient.createInternal(new AdminClientConfig(conf, true), null);
}
@Override
public void close() {
close(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
}
@Deprecated
public void close(long duration, TimeUnit unit) {
close(Duration.ofMillis(unit.toMillis(duration)));
}
public abstract void close(Duration timeout);
public ListTopicsResult listTopics() {
return listTopics(new ListTopicsOptions());
}
public abstract ListTopicsResult listTopics(ListTopicsOptions options);
}
package com.wenjie.demo.kafkaconnector.component;
import org.apache.kafka.clients.ClientDnsLookup;
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.
*/
public class AdminClientConfig extends AbstractConfig {
private static final ConfigDef CONFIG;
/**
* <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>client.dns.lookup</code>
*/
public static final String CLIENT_DNS_LOOKUP_CONFIG = CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG;
private static final String CLIENT_DNS_LOOKUP_DOC = CommonClientConfigs.CLIENT_DNS_LOOKUP_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>
*/
public static final String RETRY_BACKOFF_MS_CONFIG = CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG;
private static final String RETRY_BACKOFF_MS_DOC = "The amount of time to wait before attempting to " +
"retry a failed request. This avoids repeatedly sending requests in a tight loop under " +
"some failure scenarios.";
/**
* <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;
public static final String MAX_BUFFER_BYTES = "max.buffer.bytes";
public static final String MAX_BUFFER_BYTES_DOC = "The size of the receive buffer to use when reading data.";
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)
.define(CLIENT_DNS_LOOKUP_CONFIG,
Type.STRING,
ClientDnsLookup.DEFAULT.toString(),
in(ClientDnsLookup.DEFAULT.toString(),
ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
Importance.MEDIUM,
CLIENT_DNS_LOOKUP_DOC)
// security support
.define(SECURITY_PROTOCOL_CONFIG,
Type.STRING,
DEFAULT_SECURITY_PROTOCOL,
Importance.MEDIUM,
SECURITY_PROTOCOL_DOC)
//这里是保存默认值！！！！！!！！！！！!!
.define(MAX_BUFFER_BYTES, Type.INT, 1 << 20, Importance.MEDIUM, MAX_BUFFER_BYTES_DOC)
.withClientSslSupport()
.withClientSaslSupport();
}
@Override
protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
return CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
}
public AdminClientConfig(Map<?, ?> props) {
this(props, false);
}
protected AdminClientConfig(Map<?, ?> props, boolean doLog) {
super(CONFIG, props, doLog);
}
public static Set<String> configNames() {
return CONFIG.names();
}
public static ConfigDef configDef() {
return new ConfigDef(CONFIG);
}
public static void main(String[] args) {
System.out.println(CONFIG.toHtmlTable());
}
}
/*
* 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 com.wenjie.demo.kafkaconnector.component;
import org.apache.kafka.clients.*;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.clients.admin.internals.AdminMetadataManager;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.errors.*;
import org.apache.kafka.common.internals.KafkaFutureImpl;
import org.apache.kafka.common.message.MetadataRequestData;
import org.apache.kafka.common.metrics.*;
import org.apache.kafka.common.network.ChannelBuilder;
import org.apache.kafka.common.network.Selector;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.KafkaThread;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Predicate;
import static org.apache.kafka.common.utils.Utils.closeQuietly;
/**
* The default implementation of {@link AdminClient}. An instance of this class is created by invoking one of the
* {@code create()} methods in {@code AdminClient}. Users should not refer to this class directly.
* <p>
* The API of this class is evolving, see {@link AdminClient} for details.
*/
@InterfaceStability.Evolving
public class KafkaAdminClient extends AdminClient {
/**
* The next integer to use to name a KafkaAdminClient which the user hasn't specified an explicit name for.
*/
private static final AtomicInteger ADMIN_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
/**
* The prefix to use for the JMX metrics for this class
*/
private static final String JMX_PREFIX = "kafka.admin.client";
/**
* An invalid shutdown time which indicates that a shutdown has not yet been performed.
*/
private static final long INVALID_SHUTDOWN_TIME = -1;
/**
* Thread name prefix for admin client network thread
*/
static final String NETWORK_THREAD_PREFIX = "kafka-admin-client-thread";
private final Logger log;
/**
* The default timeout to use for an operation.
*/
private final int defaultTimeoutMs;
/**
* The name of this AdminClient instance.
*/
private final String clientId;
/**
* Provides the time.
*/
private final Time time;
/**
* The cluster metadata manager used by the KafkaClient.
*/
private final AdminMetadataManager metadataManager;
/**
* The metrics for this KafkaAdminClient.
*/
private final Metrics metrics;
/**
* The network client to use.
*/
private final KafkaClient client;
/**
* The runnable used in the service thread for this admin client.
*/
private final AdminClientRunnable runnable;
/**
* The network service thread for this admin client.
*/
private final Thread thread;
/**
* During a close operation, this is the time at which we will time out all pending operations
* and force the RPC thread to exit. If the admin client is not closing, this will be 0.
*/
private final AtomicLong hardShutdownTimeMs = new AtomicLong(INVALID_SHUTDOWN_TIME);
/**
* A factory which creates TimeoutProcessors for the RPC thread.
*/
private final TimeoutProcessorFactory timeoutProcessorFactory;
private final int maxRetries;
private final long retryBackoffMs;
/**
* Get or create a list value from a map.
*
* @param map The map to get or create the element from.
* @param key The key.
* @param <K> The key type.
* @param <V> The value type.
* @return The list value.
*/
static <K, V> List<V> getOrCreateListValue(Map<K, List<V>> map, K key) {
List<V> list = map.get(key);
if (list != null)
return list;
list = new LinkedList<>();
map.put(key, list);
return list;
}
/**
* Get the current time remaining before a deadline as an integer.
*
* @param now The current time in milliseconds.
* @param deadlineMs The deadline time in milliseconds.
* @return The time delta in milliseconds.
*/
static int calcTimeoutMsRemainingAsInt(long now, long deadlineMs) {
long deltaMs = deadlineMs - now;
if (deltaMs > Integer.MAX_VALUE)
deltaMs = Integer.MAX_VALUE;
else if (deltaMs < Integer.MIN_VALUE)
deltaMs = Integer.MIN_VALUE;
return (int) deltaMs;
}
/**
* Generate the client id based on the configuration.
*
* @param config The configuration
* @return The client id
*/
static String generateClientId(AdminClientConfig config) {
String clientId = config.getString(AdminClientConfig.CLIENT_ID_CONFIG);
if (!clientId.isEmpty())
return clientId;
return "adminclient-" + ADMIN_CLIENT_ID_SEQUENCE.getAndIncrement();
}
/**
* Get the deadline for a particular call.
*
* @param now The current time in milliseconds.
* @param optionTimeoutMs The timeout option given by the user.
* @return The deadline in milliseconds.
*/
private long calcDeadlineMs(long now, Integer optionTimeoutMs) {
if (optionTimeoutMs != null)
return now + Math.max(0, optionTimeoutMs);
return now + defaultTimeoutMs;
}
/**
* Pretty-print an exception.
*
* @param throwable The exception.
* @return A compact human-readable string.
*/
static String prettyPrintException(Throwable throwable) {
if (throwable == null)
return "Null exception.";
if (throwable.getMessage() != null) {
return throwable.getClass().getSimpleName() + ": " + throwable.getMessage();
}
return throwable.getClass().getSimpleName();
}
static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcessorFactory timeoutProcessorFactory) {
Metrics metrics = null;
NetworkClient networkClient = null;
Time time = Time.SYSTEM;
String clientId = generateClientId(config);
ChannelBuilder channelBuilder = null;
Selector selector = null;
ApiVersions apiVersions = new ApiVersions();
LogContext logContext = createLogContext(clientId);
try {
// Since we only request node information, it's safe to pass true for allowAutoTopicCreation (and it
// simplifies communication with older brokers)
AdminMetadataManager metadataManager = new AdminMetadataManager(logContext,
config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),
config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG));
List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(
config.getList(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG),
config.getString(AdminClientConfig.CLIENT_DNS_LOOKUP_CONFIG));
metadataManager.update(Cluster.bootstrap(addresses), time.milliseconds());
List<MetricsReporter> reporters = config.getConfiguredInstances(AdminClientConfig.METRIC_REPORTER_CLASSES_CONFIG,
MetricsReporter.class,
Collections.singletonMap(AdminClientConfig.CLIENT_ID_CONFIG, clientId));
Map<String, String> metricTags = Collections.singletonMap("client-id", clientId);
MetricConfig metricConfig = new MetricConfig().samples(config.getInt(AdminClientConfig.METRICS_NUM_SAMPLES_CONFIG))
.timeWindow(config.getLong(AdminClientConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
.recordLevel(Sensor.RecordingLevel.forName(config.getString(AdminClientConfig.METRICS_RECORDING_LEVEL_CONFIG)))
.tags(metricTags);
reporters.add(new JmxReporter(JMX_PREFIX));
metrics = new Metrics(metricConfig, reporters, time);
String metricGrpPrefix = "admin-client";
channelBuilder = ClientUtils.createChannelBuilder(config, time);
//注意这里！！！！！！！！！！！
selector = new Selector(config.getInt(AdminClientConfig.MAX_BUFFER_BYTES), config.getLong(AdminClientConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
metrics, time, metricGrpPrefix, Collections.emptyMap(), true, channelBuilder, logContext);
networkClient = new NetworkClient(
selector,
metadataManager.updater(),
clientId,
1,
config.getLong(AdminClientConfig.RECONNECT_BACKOFF_MS_CONFIG),
config.getLong(AdminClientConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG),
config.getInt(AdminClientConfig.SEND_BUFFER_CONFIG),
config.getInt(AdminClientConfig.RECEIVE_BUFFER_CONFIG),
(int) TimeUnit.HOURS.toMillis(1),
ClientDnsLookup.forConfig(config.getString(AdminClientConfig.CLIENT_DNS_LOOKUP_CONFIG)),
time,
true,
apiVersions,
logContext);
return new KafkaAdminClient(config, clientId, time, metadataManager, metrics, networkClient,
timeoutProcessorFactory, logContext);
} catch (Throwable exc) {
closeQuietly(metrics, "Metrics");
closeQuietly(networkClient, "NetworkClient");
closeQuietly(selector, "Selector");
closeQuietly(channelBuilder, "ChannelBuilder");
throw new KafkaException("Failed to create new KafkaAdminClient", exc);
}
}
static LogContext createLogContext(String clientId) {
return new LogContext("[AdminClient clientId=" + clientId + "] ");
}
private KafkaAdminClient(AdminClientConfig config,
String clientId,
Time time,
AdminMetadataManager metadataManager,
Metrics metrics,
KafkaClient client,
TimeoutProcessorFactory timeoutProcessorFactory,
LogContext logContext) {
this.defaultTimeoutMs = config.getInt(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG);
this.clientId = clientId;
this.log = logContext.logger(KafkaAdminClient.class);
this.time = time;
this.metadataManager = metadataManager;
this.metrics = metrics;
this.client = client;
this.runnable = new AdminClientRunnable();
String threadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.thread = new KafkaThread(threadName, runnable, true);
this.timeoutProcessorFactory = (timeoutProcessorFactory == null) ?
new TimeoutProcessorFactory() : timeoutProcessorFactory;
this.maxRetries = config.getInt(AdminClientConfig.RETRIES_CONFIG);
this.retryBackoffMs = config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG);
config.logUnused();
AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());
log.debug("Kafka admin client initialized");
thread.start();
}
@Override
public void close(Duration timeout) {
long waitTimeMs = timeout.toMillis();
if (waitTimeMs < 0)
throw new IllegalArgumentException("The timeout cannot be negative.");
waitTimeMs = Math.min(TimeUnit.DAYS.toMillis(365), waitTimeMs); // Limit the timeout to a year.
long now = time.milliseconds();
long newHardShutdownTimeMs = now + waitTimeMs;
long prev = INVALID_SHUTDOWN_TIME;
while (true) {
if (hardShutdownTimeMs.compareAndSet(prev, newHardShutdownTimeMs)) {
if (prev == INVALID_SHUTDOWN_TIME) {
log.debug("Initiating close operation.");
} else {
log.debug("Moving hard shutdown time forward.");
}
client.wakeup(); // Wake the thread, if it is blocked inside poll().
break;
}
prev = hardShutdownTimeMs.get();
if (prev < newHardShutdownTimeMs) {
log.debug("Hard shutdown time is already earlier than requested.");
newHardShutdownTimeMs = prev;
break;
}
}
if (log.isDebugEnabled()) {
long deltaMs = Math.max(0, newHardShutdownTimeMs - time.milliseconds());
log.debug("Waiting for the I/O thread to exit. Hard shutdown in {} ms.", deltaMs);
}
try {
// Wait for the thread to be joined.
thread.join();
AppInfoParser.unregisterAppInfo(JMX_PREFIX, clientId, metrics);
log.debug("Kafka admin client closed.");
} catch (InterruptedException e) {
log.debug("Interrupted while joining I/O thread", e);
Thread.currentThread().interrupt();
}
}
/**
* An interface for providing a node for a call.
*/
private interface NodeProvider {
Node provide();
}
private class MetadataUpdateNodeIdProvider implements NodeProvider {
@Override
public Node provide() {
return client.leastLoadedNode(time.milliseconds());
}
}
/**
* Provides the least loaded node.
*/
private class LeastLoadedNodeProvider implements NodeProvider {
@Override
public Node provide() {
if (metadataManager.isReady()) {
// This may return null if all nodes are busy.
// In that case, we will postpone node assignment.
return client.leastLoadedNode(time.milliseconds());
}
metadataManager.requestUpdate();
return null;
}
}
abstract class Call {
private final boolean internal;
private final String callName;
private final long deadlineMs;
private final NodeProvider nodeProvider;
private int tries = 0;
private boolean aborted = false;
private Node curNode = null;
private long nextAllowedTryMs = 0;
Call(boolean internal, String callName, long deadlineMs, NodeProvider nodeProvider) {
this.internal = internal;
this.callName = callName;
this.deadlineMs = deadlineMs;
this.nodeProvider = nodeProvider;
}
Call(String callName, long deadlineMs, NodeProvider nodeProvider) {
this(false, callName, deadlineMs, nodeProvider);
}
protected Node curNode() {
return curNode;
}
/**
* Handle a failure.
* <p>
* Depending on what the exception is and how many times we have already tried, we may choose to
* fail the Call, or retry it. It is important to print the stack traces here in some cases,
* since they are not necessarily preserved in ApiVersionException objects.
*
* @param now The current time in milliseconds.
* @param throwable The failure exception.
*/
final void fail(long now, Throwable throwable) {
if (aborted) {
// If the call was aborted while in flight due to a timeout, deliver a
// TimeoutException. In this case, we do not get any more retries - the call has
// failed. We increment tries anyway in order to display an accurate log message.
tries++;
if (log.isDebugEnabled()) {
log.debug("{} aborted at {} after {} attempt(s)", this, now, tries,
new Exception(prettyPrintException(throwable)));
}
handleFailure(new TimeoutException("Aborted due to timeout."));
return;
}
// If this is an UnsupportedVersionException that we can retry, do so. Note that a
// protocol downgrade will not count against the total number of retries we get for
// this RPC. That is why 'tries' is not incremented.
if ((throwable instanceof UnsupportedVersionException) &&
handleUnsupportedVersionException((UnsupportedVersionException) throwable)) {
log.debug("{} attempting protocol downgrade and then retry.", this);
runnable.enqueue(this, now);
return;
}
tries++;
nextAllowedTryMs = now + retryBackoffMs;
// If the call has timed out, fail.
if (calcTimeoutMsRemainingAsInt(now, deadlineMs) < 0) {
if (log.isDebugEnabled()) {
log.debug("{} timed out at {} after {} attempt(s)", this, now, tries,
new Exception(prettyPrintException(throwable)));
}
handleFailure(throwable);
return;
}
// If the exception is not retryable, fail.
if (!(throwable instanceof RetriableException)) {
if (log.isDebugEnabled()) {
log.debug("{} failed with non-retriable exception after {} attempt(s)", this, tries,
new Exception(prettyPrintException(throwable)));
}
handleFailure(throwable);
return;
}
// If we are out of retries, fail.
if (tries > maxRetries) {
if (log.isDebugEnabled()) {
log.debug("{} failed after {} attempt(s)", this, tries,
new Exception(prettyPrintException(throwable)));
}
handleFailure(throwable);
return;
}
if (log.isDebugEnabled()) {
log.debug("{} failed: {}. Beginning retry #{}",
this, prettyPrintException(throwable), tries);
}
runnable.enqueue(this, now);
}
abstract AbstractRequest.Builder createRequest(int timeoutMs);
abstract void handleResponse(AbstractResponse abstractResponse);
abstract void handleFailure(Throwable throwable);
boolean handleUnsupportedVersionException(UnsupportedVersionException exception) {
return false;
}
@Override
public String toString() {
return "Call(callName=" + callName + ", deadlineMs=" + deadlineMs + ")";
}
public boolean isInternal() {
return internal;
}
}
static class TimeoutProcessorFactory {
TimeoutProcessor create(long now) {
return new TimeoutProcessor(now);
}
}
static class TimeoutProcessor {
private final long now;
private int nextTimeoutMs;
TimeoutProcessor(long now) {
this.now = now;
this.nextTimeoutMs = Integer.MAX_VALUE;
}
int handleTimeouts(Collection<Call> calls, String msg) {
int numTimedOut = 0;
for (Iterator<Call> iter = calls.iterator(); iter.hasNext(); ) {
Call call = iter.next();
int remainingMs = calcTimeoutMsRemainingAsInt(now, call.deadlineMs);
if (remainingMs < 0) {
call.fail(now, new TimeoutException(msg));
iter.remove();
numTimedOut++;
} else {
nextTimeoutMs = Math.min(nextTimeoutMs, remainingMs);
}
}
return numTimedOut;
}
boolean callHasExpired(Call call) {
int remainingMs = calcTimeoutMsRemainingAsInt(now, call.deadlineMs);
if (remainingMs < 0)
return true;
nextTimeoutMs = Math.min(nextTimeoutMs, remainingMs);
return false;
}
int nextTimeoutMs() {
return nextTimeoutMs;
}
}
private final class AdminClientRunnable implements Runnable {
private final ArrayList<Call> pendingCalls = new ArrayList<>();
private final Map<Node, List<Call>> callsToSend = new HashMap<>();
private final Map<String, List<Call>> callsInFlight = new HashMap<>();
private final Map<Integer, Call> correlationIdToCalls = new HashMap<>();
private List<Call> newCalls = new LinkedList<>();
private void timeoutPendingCalls(TimeoutProcessor processor) {
int numTimedOut = processor.handleTimeouts(pendingCalls, "Timed out waiting for a node assignment.");
if (numTimedOut > 0)
log.debug("Timed out {} pending calls.", numTimedOut);
}
private int timeoutCallsToSend(TimeoutProcessor processor) {
int numTimedOut = 0;
for (List<Call> callList : callsToSend.values()) {
numTimedOut += processor.handleTimeouts(callList,
"Timed out waiting to send the call.");
}
if (numTimedOut > 0)
log.debug("Timed out {} call(s) with assigned nodes.", numTimedOut);
return numTimedOut;
}
private synchronized void drainNewCalls() {
if (!newCalls.isEmpty()) {
pendingCalls.addAll(newCalls);
newCalls.clear();
}
}
private long maybeDrainPendingCalls(long now) {
long pollTimeout = Long.MAX_VALUE;
log.trace("Trying to choose nodes for {} at {}", pendingCalls, now);
Iterator<Call> pendingIter = pendingCalls.iterator();
while (pendingIter.hasNext()) {
Call call = pendingIter.next();
// If the call is being retried, await the proper backoff before finding the node
if (now < call.nextAllowedTryMs) {
pollTimeout = Math.min(pollTimeout, call.nextAllowedTryMs - now);
} else if (maybeDrainPendingCall(call, now)) {
pendingIter.remove();
}
}
return pollTimeout;
}
private boolean maybeDrainPendingCall(Call call, long now) {
try {
Node node = call.nodeProvider.provide();
if (node != null) {
log.trace("Assigned {} to node {}", call, node);
call.curNode = node;
getOrCreateListValue(callsToSend, node).add(call);
return true;
} else {
log.trace("Unable to assign {} to a node.", call);
return false;
}
} catch (Throwable t) {
// Handle authentication errors while choosing nodes.
log.debug("Unable to choose node for {}", call, t);
call.fail(now, t);
return true;
}
}
private long sendEligibleCalls(long now) {
long pollTimeout = Long.MAX_VALUE;
for (Iterator<Map.Entry<Node, List<Call>>> iter = callsToSend.entrySet().iterator(); iter.hasNext(); ) {
Map.Entry<Node, List<Call>> entry = iter.next();
List<Call> calls = entry.getValue();
if (calls.isEmpty()) {
iter.remove();
continue;
}
Node node = entry.getKey();
if (!client.ready(node, now)) {
long nodeTimeout = client.pollDelayMs(node, now);
pollTimeout = Math.min(pollTimeout, nodeTimeout);
log.trace("Client is not ready to send to {}. Must delay {} ms", node, nodeTimeout);
continue;
}
Call call = calls.remove(0);
int timeoutMs = calcTimeoutMsRemainingAsInt(now, call.deadlineMs);
AbstractRequest.Builder<?> requestBuilder;
try {
requestBuilder = call.createRequest(timeoutMs);
} catch (Throwable throwable) {
call.fail(now, new KafkaException(String.format(
"Internal error sending %s to %s.", call.callName, node)));
continue;
}
ClientRequest clientRequest = client.newClientRequest(node.idString(), requestBuilder, now, true);
log.trace("Sending {} to {}. correlationId={}", requestBuilder, node, clientRequest.correlationId());
client.send(clientRequest, now);
getOrCreateListValue(callsInFlight, node.idString()).add(call);
correlationIdToCalls.put(clientRequest.correlationId(), call);
}
return pollTimeout;
}
private void timeoutCallsInFlight(TimeoutProcessor processor) {
int numTimedOut = 0;
for (Map.Entry<String, List<Call>> entry : callsInFlight.entrySet()) {
List<Call> contexts = entry.getValue();
if (contexts.isEmpty())
continue;
String nodeId = entry.getKey();
// We assume that the first element in the list is the earliest. So it should be the
// only one we need to check the timeout for.
Call call = contexts.get(0);
if (processor.callHasExpired(call)) {
if (call.aborted) {
log.warn("Aborted call {} is still in callsInFlight.", call);
} else {
log.debug("Closing connection to {} to time out {}", nodeId, call);
call.aborted = true;
client.disconnect(nodeId);
numTimedOut++;
// We don't remove anything from the callsInFlight data structure. Because the connection
// has been closed, the calls should be returned by the next client#poll(),
// and handled at that point.
}
}
}
if (numTimedOut > 0)
log.debug("Timed out {} call(s) in flight.", numTimedOut);
}
/**
* Handle responses from the server.
*
* @param now The current time in milliseconds.
* @param responses The latest responses from KafkaClient.
**/
private void handleResponses(long now, List<ClientResponse> responses) {
for (ClientResponse response : responses) {
int correlationId = response.requestHeader().correlationId();
Call call = correlationIdToCalls.get(correlationId);
if (call == null) {
// If the server returns information about a correlation ID we didn't use yet,
// an internal server error has occurred. Close the connection and log an error message.
log.error("Internal server error on {}: server returned information about unknown " +
"correlation ID {}, requestHeader = {}", response.destination(), correlationId,
response.requestHeader());
client.disconnect(response.destination());
continue;
}
// Stop tracking this call.
correlationIdToCalls.remove(correlationId);
List<Call> calls = callsInFlight.get(response.destination());
if ((calls == null) || (!calls.remove(call))) {
log.error("Internal server error on {}: ignoring call {} in correlationIdToCall " +
"that did not exist in callsInFlight", response.destination(), call);
continue;
}
// Handle the result of the call. This may involve retrying the call, if we got a
// retryible exception.
if (response.versionMismatch() != null) {
call.fail(now, response.versionMismatch());
} else if (response.wasDisconnected()) {
AuthenticationException authException = client.authenticationException(call.curNode());
if (authException != null) {
call.fail(now, authException);
} else {
call.fail(now, new DisconnectException(String.format(
"Cancelled %s request with correlation id %s due to node %s being disconnected",
call.callName, correlationId, response.destination())));
}
} else {
try {
call.handleResponse(response.responseBody());
if (log.isTraceEnabled())
log.trace("{} got response {}", call,
response.responseBody().toString(response.requestHeader().apiVersion()));
} catch (Throwable t) {
if (log.isTraceEnabled())
log.trace("{} handleResponse failed with {}", call, prettyPrintException(t));
call.fail(now, t);
}
}
}
}
private void unassignUnsentCalls(Predicate<Node> shouldUnassign) {
for (Iterator<Map.Entry<Node, List<Call>>> iter = callsToSend.entrySet().iterator(); iter.hasNext(); ) {
Map.Entry<Node, List<Call>> entry = iter.next();
Node node = entry.getKey();
List<Call> awaitingCalls = entry.getValue();
if (awaitingCalls.isEmpty()) {
iter.remove();
} else if (shouldUnassign.test(node)) {
pendingCalls.addAll(awaitingCalls);
iter.remove();
}
}
}
private boolean hasActiveExternalCalls(Collection<Call> calls) {
for (Call call : calls) {
if (!call.isInternal()) {
return true;
}
}
return false;
}
/**
* Return true if there are currently active external calls.
*/
private boolean hasActiveExternalCalls() {
if (hasActiveExternalCalls(pendingCalls)) {
return true;
}
for (List<Call> callList : callsToSend.values()) {
if (hasActiveExternalCalls(callList)) {
return true;
}
}
return hasActiveExternalCalls(correlationIdToCalls.values());
}
private boolean threadShouldExit(long now, long curHardShutdownTimeMs) {
if (!hasActiveExternalCalls()) {
log.trace("All work has been completed, and the I/O thread is now exiting.");
return true;
}
if (now >= curHardShutdownTimeMs) {
log.info("Forcing a hard I/O thread shutdown. Requests in progress will be aborted.");
return true;
}
log.debug("Hard shutdown in {} ms.", curHardShutdownTimeMs - now);
return false;
}
@Override
public void run() {
long now = time.milliseconds();
log.trace("Thread starting");
while (true) {
// Copy newCalls into pendingCalls.
drainNewCalls();
// Check if the AdminClient thread should shut down.
long curHardShutdownTimeMs = hardShutdownTimeMs.get();
if ((curHardShutdownTimeMs != INVALID_SHUTDOWN_TIME) && threadShouldExit(now, curHardShutdownTimeMs))
break;
// Handle timeouts.
TimeoutProcessor timeoutProcessor = timeoutProcessorFactory.create(now);
timeoutPendingCalls(timeoutProcessor);
timeoutCallsToSend(timeoutProcessor);
timeoutCallsInFlight(timeoutProcessor);
long pollTimeout = Math.min(1200000, timeoutProcessor.nextTimeoutMs());
if (curHardShutdownTimeMs != INVALID_SHUTDOWN_TIME) {
pollTimeout = Math.min(pollTimeout, curHardShutdownTimeMs - now);
}
// Choose nodes for our pending calls.
pollTimeout = Math.min(pollTimeout, maybeDrainPendingCalls(now));
long metadataFetchDelayMs = metadataManager.metadataFetchDelayMs(now);
if (metadataFetchDelayMs == 0) {
metadataManager.transitionToUpdatePending(now);
Call metadataCall = makeMetadataCall(now);
// Create a new metadata fetch call and add it to the end of pendingCalls.
// Assign a node for just the new call (we handled the other pending nodes above).
if (!maybeDrainPendingCall(metadataCall, now))
pendingCalls.add(metadataCall);
}
pollTimeout = Math.min(pollTimeout, sendEligibleCalls(now));
if (metadataFetchDelayMs > 0) {
pollTimeout = Math.min(pollTimeout, metadataFetchDelayMs);
}
// Ensure that we use a small poll timeout if there are pending calls which need to be sent
if (!pendingCalls.isEmpty())
pollTimeout = Math.min(pollTimeout, retryBackoffMs);
// Wait for network responses.
log.trace("Entering KafkaClient#poll(timeout={})", pollTimeout);
List<ClientResponse> responses = client.poll(pollTimeout, now);
log.trace("KafkaClient#poll retrieved {} response(s)", responses.size());
// unassign calls to disconnected nodes
unassignUnsentCalls(client::connectionFailed);
// Update the current time and handle the latest responses.
now = time.milliseconds();
handleResponses(now, responses);
}
int numTimedOut = 0;
TimeoutProcessor timeoutProcessor = new TimeoutProcessor(Long.MAX_VALUE);
synchronized (this) {
numTimedOut += timeoutProcessor.handleTimeouts(newCalls, "The AdminClient thread has exited.");
newCalls = null;
}
numTimedOut += timeoutProcessor.handleTimeouts(pendingCalls, "The AdminClient thread has exited.");
numTimedOut += timeoutCallsToSend(timeoutProcessor);
numTimedOut += timeoutProcessor.handleTimeouts(correlationIdToCalls.values(),
"The AdminClient thread has exited.");
if (numTimedOut > 0) {
log.debug("Timed out {} remaining operation(s).", numTimedOut);
}
closeQuietly(client, "KafkaClient");
closeQuietly(metrics, "Metrics");
log.debug("Exiting AdminClientRunnable thread.");
}
void enqueue(Call call, long now) {
if (log.isDebugEnabled()) {
log.debug("Queueing {} with a timeout {} ms from now.", call, call.deadlineMs - now);
}
boolean accepted = false;
synchronized (this) {
if (newCalls != null) {
newCalls.add(call);
accepted = true;
}
}
if (accepted) {
client.wakeup(); // wake the thread if it is in poll()
} else {
log.debug("The AdminClient thread has exited. Timing out {}.", call);
call.fail(Long.MAX_VALUE, new TimeoutException("The AdminClient thread has exited."));
}
}
void call(Call call, long now) {
if (hardShutdownTimeMs.get() != INVALID_SHUTDOWN_TIME) {
log.debug("The AdminClient is not accepting new calls. Timing out {}.", call);
call.fail(Long.MAX_VALUE, new TimeoutException("The AdminClient thread is not accepting new calls."));
} else {
enqueue(call, now);
}
}
private Call makeMetadataCall(long now) {
return new Call(true, "fetchMetadata", calcDeadlineMs(now, defaultTimeoutMs),
new MetadataUpdateNodeIdProvider()) {
@Override
public AbstractRequest.Builder createRequest(int timeoutMs) {
// Since this only requests node information, it's safe to pass true
// for allowAutoTopicCreation (and it simplifies communication with
// older brokers)
return new MetadataRequest.Builder(new MetadataRequestData()
.setTopics(Collections.emptyList())
.setAllowAutoTopicCreation(true));
}
@Override
public void handleResponse(AbstractResponse abstractResponse) {
MetadataResponse response = (MetadataResponse) abstractResponse;
long now = time.milliseconds();
metadataManager.update(response.cluster(), now);
// Unassign all unsent requests after a metadata refresh to allow for a new
// destination to be selected from the new metadata
unassignUnsentCalls(node -> true);
}
@Override
public void handleFailure(Throwable e) {
metadataManager.updateFailed(e);
}
};
}
}
@Override
public ListTopicsResult listTopics(final ListTopicsOptions options) {
final KafkaFutureImpl<Map<String, TopicListing>> topicListingFuture = new KafkaFutureImpl<>();
final long now = time.milliseconds();
runnable.call(new Call("listTopics", calcDeadlineMs(now, options.timeoutMs()),
new LeastLoadedNodeProvider()) {
@Override
AbstractRequest.Builder createRequest(int timeoutMs) {
return MetadataRequest.Builder.allTopics();
}
@Override
void handleResponse(AbstractResponse abstractResponse) {
MetadataResponse response = (MetadataResponse) abstractResponse;
Map<String, TopicListing> topicListing = new HashMap<>();
for (MetadataResponse.TopicMetadata topicMetadata : response.topicMetadata()) {
String topicName = topicMetadata.topic();
boolean isInternal = topicMetadata.isInternal();
if (!topicMetadata.isInternal() || options.shouldListInternal())
topicListing.put(topicName, new TopicListing(topicName, isInternal));
}
topicListingFuture.complete(topicListing);
}
@Override
void handleFailure(Throwable throwable) {
topicListingFuture.completeExceptionally(throwable);
}
}, now);
return new ListTopicsResult(topicListingFuture);
}
}
package com.wenjie.demo.kafkaconnector.component;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
/**
* The result of the {@link AdminClient#listTopics()} call.
*
* The API of this class is evolving, see {@link AdminClient} for details.
*/
@InterfaceStability.Evolving
public class ListTopicsResult {
final KafkaFuture<Map<String, TopicListing>> future;
ListTopicsResult(KafkaFuture<Map<String, TopicListing>> future) {
this.future = future;
}
/**
* Return a future which yields a map of topic names to TopicListing objects.
*/
public KafkaFuture<Map<String, TopicListing>> namesToListings() {
return future;
}
/**
* Return a future which yields a collection of TopicListing objects.
*/
public KafkaFuture<Collection<TopicListing>> listings() {
return future.thenApply(new KafkaFuture.BaseFunction<Map<String, TopicListing>, Collection<TopicListing>>() {
@Override
public Collection<TopicListing> apply(Map<String, TopicListing> namesToDescriptions) {
return namesToDescriptions.values();
}
});
}
/**
* Return a future which yields a collection of topic names.
*/
public KafkaFuture<Set<String>> names() {
return future.thenApply(new KafkaFuture.BaseFunction<Map<String, TopicListing>, Set<String>>() {
@Override
public Set<String> apply(Map<String, TopicListing> namesToListings) {
return namesToListings.keySet();
}
});
}
}
注意：
在Kafka 0.10.2.0之前，Kafka服务器端和客户端版本之间的兼容性是“单向”的，即高版本的broker可以处理低版本client的请求。反过来，低版本的broker不能处理高版本client的请求。由于升级client要远比升级broker简单得多，因此这个限制给很多用户带来了麻烦，甚至有很多人都不愿意去升级broker版本——毕竟无downtime的情况下正确升级Kafka服务器是个不小的挑战。自0.10.2.0版本开始，社区对这个问题进行了优化，0.10.2.0之后用户可以简单地升级client端代码到这个版本就可以很容易地实现与低版本Kafka服务器的交互了。
魔乐社区（Modelers.cn) 是一个中立、公益的人工智能社区，提供人工智能工具、模型、数据的托管、展示与应用协同服务，为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作，由全产业链共同建设、共同运营、共同享有，推动国产AI生态繁荣发展。
更多推荐
4
0- 0
已为社区贡献1条内容
所有评论(0)