Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

GH-3483: Allow @Override of KafkaAdmin createAdmin() #3484

Merged
merged 4 commits into from
Oct 4, 2024
Merged

GH-3483: Allow @Override of KafkaAdmin createAdmin() #3484

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4e54723
GH-3483: Allow @Override of KafkaAdmin createAdmin()
anders-swanson Sep 13, 2024
88d49de
JavaDoc for KafkaAdmin#createAdmin method
anders-swanson Oct 3, 2024
2fd6218
Document KafkaAdmin#createAdmin method changes
anders-swanson Oct 3, 2024
bb1a50d
Merge branch 'main' into kafka-admin-create-override
anders-swanson Oct 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 14 additions & 13 deletions spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
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;
Expand Down Expand Up @@ -68,7 +69,7 @@
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
Expand Down Expand Up @@ -114,9 +115,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);
Expand Down Expand Up @@ -251,7 +252,7 @@ public void afterSingletonsInstantiated() {
public final boolean initialize() {
Collection<NewTopic> newTopics = newTopics();
if (!newTopics.isEmpty()) {
AdminClient adminClient = null;
Admin adminClient = null;
try {
adminClient = createAdmin();
}
Expand Down Expand Up @@ -347,7 +348,7 @@ protected Collection<NewTopic> newTopics() {
@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";
Expand All @@ -365,14 +366,14 @@ public String clusterId() {

@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 {
Expand All @@ -389,7 +390,7 @@ public Map<String, TopicDescription> describeTopics(String... topicNames) {
}
}

AdminClient createAdmin() {
protected Admin createAdmin() {
anders-swanson marked this conversation as resolved.
Show resolved Hide resolved
return AdminClient.create(getAdminConfig());
}

Expand All @@ -409,7 +410,7 @@ protected Map<String, Object> getAdminConfig() {
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));
Expand Down Expand Up @@ -439,7 +440,7 @@ private void addOrModifyTopicsIfNeeded(AdminClient adminClient, Collection<NewTo
}

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()))
Expand Down Expand Up @@ -484,7 +485,7 @@ private Map<ConfigResource, List<ConfigEntry>> checkTopicsForConfigMismatches(
}
}

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();
Expand Down Expand Up @@ -556,7 +557,7 @@ else if (topic.numPartitions() > topicDescription.partitions().size()) {
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);
Expand All @@ -579,7 +580,7 @@ private void addTopics(AdminClient adminClient, List<NewTopic> topicsToAdd) {
}
}

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);
Expand Down
3 changes: 2 additions & 1 deletion spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
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;
Expand Down Expand Up @@ -286,7 +287,7 @@ void nullClusterId() {
KafkaAdmin admin = new KafkaAdmin(Map.of()) {

@Override
AdminClient createAdmin() {
protected Admin createAdmin() {
return mock;
}

Expand Down