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

GH-3483: Allow Overriding KafkaAdmin#createAdmin()

Browse Source 
Fixes: #3483 

https://github.com/spring-projects/spring-kafka/issues/3483

- Refactor `KafkaAdmin` to use `Admin` interface
- Change `createAdmin()` method to protected visibility
- Update the return type to `org.apache.kafka.clients.admin.Admin`
- Modify `KafkaAdmin` to use `Admin` interface instead of `AdminClient` class
This commit is contained in:
Anders Swanson
2024-10-04 11:41:43 -07:00
committed by GitHub
parent a12edfb804
commit 8da1aa30ef
3 changed files with 31 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc
View File

@@ -46,7 +46,13 @@ For more details, see xref:kafka/sending-messages.adoc[Sending Messages] section
When using `DeadLetterPublishingRecovererFactory`, the user applications can override the `maybeLogListenerException` method to customize the logging behavior.
[[x33-customize-admin-client-in-KafkaAdmin]]
=== Customize Admin client in KafkaAdmin
When extending `KafkaAdmin`, user applications may override the `createAdmin` method to customize Admin client creation.
[[x33-customize-kafka-streams-implementation]]
=== Customizing The Implementation of Kafka Streams
When using `KafkaStreamsCustomizer` it is now possible to return a custom implementation of the `KafkaStreams` object by overriding the `initKafkaStreams` method.
When using `KafkaStreamsCustomizer` it is now possible to return a custom implementation of the `KafkaStreams` object by overriding the `initKafkaStreams` method.

34
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java
View File

@@ -36,6 +36,7 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterConfigOp;
@@ -68,7 +69,7 @@ import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* An admin that delegates to an {@link AdminClient} to create topics defined
* An admin that delegates to an {@link Admin} to create topics defined
* in the application context.
*
* @author Gary Russell
@@ -76,6 +77,7 @@ import org.springframework.util.Assert;
* @author Adrian Gygax
* @author Sanghyeok An
* @author Valentina Armenise
* @author Anders Swanson
*
* @since 1.3
*/
@@ -114,9 +116,9 @@ public class KafkaAdmin extends KafkaResourceFactory
private String clusterId;
/**
* Create an instance with an {@link AdminClient} based on the supplied
* Create an instance with an {@link Admin} based on the supplied
* configuration.
* @param config the configuration for the {@link AdminClient}.
* @param config the configuration for the {@link Admin}.
*/
public KafkaAdmin(Map<String, Object> config) {
this.configs = new HashMap<>(config);
@@ -251,7 +253,7 @@ public class KafkaAdmin extends KafkaResourceFactory
public final boolean initialize() {
Collection<NewTopic> newTopics = newTopics();
if (!newTopics.isEmpty()) {
AdminClient adminClient = null;
Admin adminClient = null;
try {
adminClient = createAdmin();
}
@@ -347,7 +349,7 @@ public class KafkaAdmin extends KafkaResourceFactory
@Nullable
public String clusterId() {
if (this.clusterId == null) {
try (AdminClient client = createAdmin()) {
try (Admin client = createAdmin()) {
this.clusterId = client.describeCluster().clusterId().get(this.operationTimeout, TimeUnit.SECONDS);
if (this.clusterId == null) {
this.clusterId = "null";
@@ -365,14 +367,14 @@ public class KafkaAdmin extends KafkaResourceFactory
@Override
public void createOrModifyTopics(NewTopic... topics) {
try (AdminClient client = createAdmin()) {
try (Admin client = createAdmin()) {
addOrModifyTopicsIfNeeded(client, Arrays.asList(topics));
}
}
@Override
public Map<String, TopicDescription> describeTopics(String... topicNames) {
try (AdminClient admin = createAdmin()) {
try (Admin admin = createAdmin()) {
Map<String, TopicDescription> results = new HashMap<>();
DescribeTopicsResult topics = admin.describeTopics(Arrays.asList(topicNames));
try {
@@ -389,7 +391,13 @@ public class KafkaAdmin extends KafkaResourceFactory
}
}
AdminClient createAdmin() {
/**
* Creates a new {@link Admin} client instance using the {@link AdminClient} class.
* @return the new {@link Admin} client instance.
* @since 3.3.0
* @see AdminClient#create(Map)
*/
protected Admin createAdmin() {
return AdminClient.create(getAdminConfig());
}
@@ -409,7 +417,7 @@ public class KafkaAdmin extends KafkaResourceFactory
return configs2;
}
private void addOrModifyTopicsIfNeeded(AdminClient adminClient, Collection<NewTopic> topics) {
private void addOrModifyTopicsIfNeeded(Admin adminClient, Collection<NewTopic> topics) {
if (!topics.isEmpty()) {
Map<String, NewTopic> topicNameToTopic = new HashMap<>();
topics.forEach(t -> topicNameToTopic.compute(t.name(), (k, v) -> t));
@@ -439,7 +447,7 @@ public class KafkaAdmin extends KafkaResourceFactory
}
private Map<ConfigResource, List<ConfigEntry>> checkTopicsForConfigMismatches(
AdminClient adminClient, Collection<NewTopic> topics) {
Admin adminClient, Collection<NewTopic> topics) {
List<ConfigResource> configResources = topics.stream()
.map(topic -> new ConfigResource(Type.TOPIC, topic.name()))
@@ -484,7 +492,7 @@ public class KafkaAdmin extends KafkaResourceFactory
}
}
private void adjustConfigMismatches(AdminClient adminClient, Collection<NewTopic> topics,
private void adjustConfigMismatches(Admin adminClient, Collection<NewTopic> topics,
Map<ConfigResource, List<ConfigEntry>> mismatchingConfigs) {
for (Map.Entry<ConfigResource, List<ConfigEntry>> mismatchingConfigsOfTopic : mismatchingConfigs.entrySet()) {
ConfigResource topicConfigResource = mismatchingConfigsOfTopic.getKey();
@@ -556,7 +564,7 @@ public class KafkaAdmin extends KafkaResourceFactory
return topicsToModify;
}
private void addTopics(AdminClient adminClient, List<NewTopic> topicsToAdd) {
private void addTopics(Admin adminClient, List<NewTopic> topicsToAdd) {
CreateTopicsResult topicResults = adminClient.createTopics(topicsToAdd);
try {
topicResults.all().get(this.operationTimeout, TimeUnit.SECONDS);
@@ -579,7 +587,7 @@ public class KafkaAdmin extends KafkaResourceFactory
}
}
private void createMissingPartitions(AdminClient adminClient, Map<String, NewPartitions> topicsToModify) {
private void createMissingPartitions(Admin adminClient, Map<String, NewPartitions> topicsToModify) {
CreatePartitionsResult partitionsResult = adminClient.createPartitions(topicsToModify);
try {
partitionsResult.all().get(this.operationTimeout, TimeUnit.SECONDS);

4
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java
View File

@@ -34,6 +34,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterConfigOp;
@@ -69,6 +70,7 @@ import org.springframework.util.ReflectionUtils;
/**
* @author Gary Russell
* @author Adrian Gygax
* @author Anders Swanson
*
* @since 1.3
*/
@@ -286,7 +288,7 @@ public class KafkaAdminTests {
KafkaAdmin admin = new KafkaAdmin(Map.of()) {
@Override
AdminClient createAdmin() {
protected Admin createAdmin() {
return mock;
}