OmniaGM commented on code in PR #12577:
URL: https://github.com/apache/kafka/pull/12577#discussion_r991566688
##########
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.",
+ topicName);
+ }
+ if (cause instanceof ClusterAuthorizationException) {
+ log.debug("Not authorized to create topic '{}' upon the
brokers." +
+ " Falling back to assume topic(s) exist or
will be auto-created by the broker.",
+ topicName);
+ }
+ if (cause instanceof InvalidConfigurationException) {
+ throw new ConnectException("Unable to create topic '" +
topicName + "': " + cause.getMessage(),
+ cause);
+ }
+ if (cause instanceof TimeoutException) {
+ // Timed out waiting for the operation to complete
Review Comment:
Same here as well, this comment is added in `TopicAdmin.createOrFindTopics`
so I kept it.
--
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]
