Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Deprecate ZkClient in embedded broker

Browse Source 
Deprecate `ZkClient` in favor of Kafka's `ZooKeeperClient`
in preparation for the 2.4 clients, where this class is removed.

Also only create the client on-demand because it is not used internally
by the embedded broker.
This commit is contained in:
Gary Russell
2019-10-28 15:56:02 -04:00
committed by Artem Bilan
parent ec446d4da4
commit 9eadf5bca8
1 changed files with 42 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java
View File

@@ -70,6 +70,7 @@ import kafka.utils.CoreUtils;
import kafka.utils.TestUtils;
import kafka.utils.ZKStringSerializer$;
import kafka.zk.ZkFourLetterWords;
import kafka.zookeeper.ZooKeeperClient;
/**
* An embedded Kafka Broker(s) and Zookeeper manager.
@@ -104,6 +105,10 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
private static final Duration DEFAULT_ADMIN_TIMEOUT = Duration.ofSeconds(10);
private static final int ZK_CONNECTION_TIMEOUT = 6000;
private static final int ZK_SESSION_TIMEOUT = 6000;
private final int count;
private final boolean controlledShutdown;
@@ -118,8 +123,6 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
private EmbeddedZookeeper zookeeper;
private ZkClient zookeeperClient;
private String zkConnect;
private int zkPort;
@@ -130,6 +133,10 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
private String brokerListProperty;
private volatile ZooKeeperClient zooKeeperClient;
private volatile ZkClient zkClient;
public EmbeddedKafkaBroker(int count) {
this(count, false);
}
@@ -257,12 +264,7 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
catch (IOException | InterruptedException e) {
throw new IllegalStateException("Failed to create embedded Zookeeper", e);
}
int zkConnectionTimeout = 6000; // NOSONAR magic #
int zkSessionTimeout = 6000; // NOSONAR magic #
this.zkConnect = LOOPBACK + ":" + this.zookeeper.getPort();
this.zookeeperClient = new ZkClient(this.zkConnect, zkSessionTimeout, zkConnectionTimeout,
ZKStringSerializer$.MODULE$);
this.kafkaServers.clear();
for (int i = 0; i < this.count; i++) {
Properties brokerConfigProperties = createBrokerProperties(i);
@@ -396,7 +398,14 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
}
}
try {
this.zookeeperClient.close();
synchronized (this) {
if (this.zooKeeperClient != null) {
this.zooKeeperClient.close();
}
if (this.zkClient != null) {
this.zkClient.close();
}
}
}
catch (ZkInterruptedException e) {
// do nothing
@@ -426,8 +435,31 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
return this.zookeeper;
}
public ZkClient getZkClient() {
return this.zookeeperClient;
/**
* Return the ZkClient.
* @return the client.
* @deprecated in favor of {@link #getZooKeeperClient()}.
*/
@Deprecated
public synchronized ZkClient getZkClient() {
if (this.zkClient == null) {
this.zkClient = new ZkClient(this.zkConnect, ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT,
ZKStringSerializer$.MODULE$);
}
return this.zkClient;
}
/**
* Return the ZooKeeperClient.
* @return the client.
* @since 2.3.2
*/
public synchronized ZooKeeperClient getZooKeeperClient() {
if (this.zooKeeperClient == null) {
this.zooKeeperClient = new ZooKeeperClient(this.zkConnect, ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT,
1, Time.SYSTEM, "embeddedKafkaZK", "embeddedKafkaZK");
}
return this.zooKeeperClient;
}
public String getZookeeperConnectionString() {