OmniaGM commented on code in PR #12577:
URL: https://github.com/apache/kafka/pull/12577#discussion_r991564622
##########
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);
+ }
+ if (cause instanceof UnsupportedVersionException) {
+ log.debug("Unable to create topic(s) '{}' since the brokers do
not support the CreateTopics API." +
Review Comment:
fixed.
##########
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);
+ }
+ if (cause instanceof UnsupportedVersionException) {
+ log.debug("Unable to create topic(s) '{}' since the brokers do
not support the CreateTopics API." +
+ " Falling back to assume topic exist or will
be auto-created by the broker.",
Review Comment:
fixed.
--
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]
