MINOR: Fix AuthorizationErrorTest (#1304)

Fix AuthorizationErrorTest failures due to the changes in https://github.com/apache/kafka/pull/17224/files

Author: omkreddy

kafka-rest/src/test/java/io/confluent/kafkarest/integration/AuthorizationErrorTest.java

@@ -39,7 +39,13 @@
 import javax.ws.rs.core.Response;
 import kafka.security.authorizer.AclAuthorizer;
 import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.junit.jupiter.api.AfterEach;
@@ -151,18 +157,36 @@ public void testConsumerRequest(String quorum) {
     // test without acls
     verifySubscribeToTopic(true);
     // add acls
-    SecureTestUtils.setConsumerAcls(zkConnect, TOPIC_NAME, USERNAME, CONSUMER_GROUP);
+    setConsumerAcls();
     verifySubscribeToTopic(false);
   }
 
+  private void setConsumerAcls() {
+    AclBinding topicAcl =
+        new AclBinding(
+            new ResourcePattern(ResourceType.TOPIC, TOPIC_NAME, PatternType.LITERAL),
+            new AccessControlEntry(
+                "User:" + USERNAME, "*", AclOperation.READ, AclPermissionType.ALLOW));
+    AclBinding groupAcl =
+        new AclBinding(
+            new ResourcePattern(ResourceType.GROUP, CONSUMER_GROUP, PatternType.LITERAL),
+            new AccessControlEntry(
+                "User:" + USERNAME, "*", AclOperation.READ, AclPermissionType.ALLOW));
+    try {
+      createAcls(Arrays.asList(topicAcl, groupAcl), adminProperties());
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
   @ValueSource(strings = {"zk"})
   public void testProducerAuthorization(String quorum) {
     BinaryTopicProduceRequest request = BinaryTopicProduceRequest.create(topicRecords);
     // test without any acls
     testProduceToAuthorizationError(TOPIC_NAME, request);
     // add acls
-    SecureTestUtils.setProduceAcls(zkConnect, TOPIC_NAME, USERNAME);
+    setProduceAcls();
     testProduceToTopic(
         TOPIC_NAME,
         request,
@@ -173,6 +197,19 @@ public void testProducerAuthorization(String quorum) {
         request.toProduceRequest().getRecords());
   }
 
+  private void setProduceAcls() {
+    AclBinding topicAcl =
+        new AclBinding(
+            new ResourcePattern(ResourceType.TOPIC, TOPIC_NAME, PatternType.LITERAL),
+            new AccessControlEntry(
+                "User:" + USERNAME, "*", AclOperation.WRITE, AclPermissionType.ALLOW));
+    try {
+      createAcls(Arrays.asList(topicAcl), adminProperties());
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   private void verifySubscribeToTopic(boolean expectFailure) {
     Response createResponse = createConsumerInstance(CONSUMER_GROUP);
     assertOKResponse(createResponse, Versions.KAFKA_V2_JSON);
@@ -224,7 +261,30 @@ protected SecurityProtocol getBrokerSecurityProtocol() {
   @Override
   protected void setupAcls() {
     // to allow plaintext consumer
-    SecureTestUtils.setConsumerAcls(zkConnect, TOPIC_NAME, KafkaPrincipal.ANONYMOUS.getName(), "*");
+    AclBinding topicAcl =
+        new AclBinding(
+            new ResourcePattern(ResourceType.TOPIC, TOPIC_NAME, PatternType.LITERAL),
+            new AccessControlEntry(
+                "User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW));
+    AclBinding groupAcl =
+        new AclBinding(
+            new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
+            new AccessControlEntry(
+                "User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW));
+    try {
+      createAcls(Arrays.asList(topicAcl, groupAcl), adminProperties());
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private Properties adminProperties() {
+    Properties adminProperties = new Properties();
+    adminProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+    adminProperties.put("security.protocol", "SASL_PLAINTEXT");
+    adminProperties.setProperty("sasl.mechanism", "PLAIN");
+    adminProperties.put("sasl.jaas.config", createPlainLoginModule("admin", "admin-secret"));
+    return adminProperties;
   }
 
   @AfterEach

kafka-rest/src/test/java/io/confluent/kafkarest/integration/ClusterTestHarness.java

@@ -39,6 +39,7 @@
 import java.net.URISyntaxException;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -50,6 +51,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -77,6 +79,7 @@
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.acl.AclBinding;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
@@ -622,6 +625,13 @@ protected final void createTopic(
     }
   }
 
+  protected final void createAcls(Collection<AclBinding> acls, Properties adminProperties)
+      throws Exception {
+    try (AdminClient admin = AdminClient.create(adminProperties)) {
+      admin.createAcls(acls).all().get(60, TimeUnit.SECONDS);
+    }
+  }
+
   @SuppressWarnings({"unchecked"})
   private static Map<Object, Seq<Object>> convertReplicasAssignmentToScalaCompatibleType(
       Optional<Map<Integer, List<Integer>>> replicasAssignments) {