+ private String[] getTopicPermission(String topicName, AclOperation aclOperation) {
+
+ String namspace = topicName.substring(0, topicName.lastIndexOf("."));
+ String[] permission = new String[3];
+ if (TOPIC_READ_WRITE_DESCRIBE_OPERATIONS.contains(aclOperation)) {
+ permission[0] = namspace + ".topic";
+ String instancePart = (System.getenv("pubSubInstPart") != null) ? System.getenv("pubSubInstPart")
+ : ".topic";
+ permission[1] = instancePart + topicName;
+
+ if (aclOperation.equals(AclOperation.WRITE)) {
+ permission[2] = "pub";
+ } else if (aclOperation.equals(AclOperation.READ)) {
+ permission[2] = "sub";
+
+ } else if (TOPIC_DESCRIBE_OPERATIONS.contains(aclOperation)) {
+ permission[2] = "view";
+
+ }
+ } else if (aclOperation.equals(AclOperation.DELETE)) {
+ permission = new String(System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|destroy").split("\\|");
+
+ } else if (TOPIC_ADMIN_OPERATIONS.contains(aclOperation)) {
+ permission = new String(System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|create").split("\\|");
+ }
+
+ return permission;
+ }
+
+ private String[] getAdminPermission() {
+
+ if (adminPermission[0] == null) {
+ adminPermission[0] = System.getProperty("namespace") + ".kafka.access";
+ adminPermission[1] = "*";
+ adminPermission[2] = "*";
+ }
+
+ return adminPermission;
+ }
+
+ private String[] getPermission(AclOperation aclOperation, String resource, String topicName) {
+ String[] permission = new String[3];
+ switch (aclOperation) {
+
+ case ALTER:
+ case ALTER_CONFIGS:
+ case CREATE:
+ case DELETE:
+ if (resource.equals("Topic")) {
+ permission = getTopicPermission(topicName, aclOperation);
+ } else if (resource.equals("Cluster")) {
+ permission = getAdminPermission();
+ }
+ break;
+ case DESCRIBE_CONFIGS:
+ case READ:
+ case WRITE:
+ if (resource.equals("Topic")) {
+ permission = getTopicPermission(topicName, aclOperation);
+ }
+ break;
+ case IDEMPOTENT_WRITE:
+ if (resource.equals("Cluster")) {
+ permission = getAdminPermission();
+ }
+ break;
+ default:
+ break;
+
+ }
+
+ return permission;
+
+ }
+