OmniaGM commented on code in PR #12577:
URL: https://github.com/apache/kafka/pull/12577#discussion_r991566108
##########
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) {
Review Comment:
This is the same error message in `TopicAdmin.createOrFindTopics` as I am
trying to keep the same error handling behaviour and logs for customers while
switching to use ForwardingAdmin instead of `Admin`, `AdminClient` or
`TopicAdmin`
--
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]
