OmniaGM commented on code in PR #12577:
URL: https://github.com/apache/kafka/pull/12577#discussion_r991564465
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorUtils.java:
##########
@@ -98,19 +112,51 @@ static Pattern compilePatternList(String fields) {
return compilePatternList(Arrays.asList(fields.split("\\W*,\\W*")));
}
- static void createCompactedTopic(String topicName, short partitions, short
replicationFactor, Map<String, Object> adminProps) {
+ static void createCompactedTopic(String topicName, short partitions, short
replicationFactor, ForwardingAdmin forwardingAdmin) {
NewTopic topicDescription = TopicAdmin.defineTopic(topicName).
compacted().
partitions(partitions).
replicationFactor(replicationFactor).
build();
- try (TopicAdmin admin = new TopicAdmin(adminProps)) {
- admin.createTopics(topicDescription);
+ CreateTopicsOptions args = new
CreateTopicsOptions().validateOnly(false);
+ try {
+ forwardingAdmin.createTopics(singleton(topicDescription),
args).values().get(topicName).get();
+ log.info("Created topic", topicName);
+ } catch (InterruptedException e) {
+ Thread.interrupted();
+ throw new ConnectException("Interrupted while attempting to
create/find topic '" + topicName + "'", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof TopicExistsException) {
+ log.debug("Found existing topic '{}'", topicName);
Review Comment:
updated the log message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
