Add handling for GroupAuthorizationException

Add integration test for error classification

ksqldb-engine/src/main/java/io/confluent/ksql/query/GroupAuthorizationClassifier.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2021 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"; you may not use
+ * this file except in compliance with the License. You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.query;
+
+import io.confluent.ksql.query.QueryError.Type;
+import java.util.Objects;
+import org.apache.kafka.common.errors.GroupAuthorizationException;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@code GroupAuthorizationClassifier} classifies missing consumer group ACLs as user error
+ */
+public class GroupAuthorizationClassifier implements QueryErrorClassifier {
+
+  private static final Logger LOG = LoggerFactory.getLogger(GroupAuthorizationClassifier.class);
+
+  private final String queryId;
+
+  public GroupAuthorizationClassifier(final String queryId) {
+    this.queryId = Objects.requireNonNull(queryId, "queryId");
+  }
+
+  @Override
+  public Type classify(final Throwable e) {
+    final Type type = e instanceof GroupAuthorizationException ? Type.USER : Type.UNKNOWN;
+
+    if (type == Type.USER) {
+      LOG.info(
+          "Classified error as USER error based on missing consumer groups access rights. Query ID: {} Exception: {}",
+          queryId,
+          e);
+    }
+
+    return type;
+  }
+
+}

ksqldb-engine/src/main/java/io/confluent/ksql/query/MissingTopicClassifier.java

@@ -22,9 +22,7 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * {@code MissingTopicClassifier} classifies errors by querying the broker
- * to check that all topics that the query relies on being accessible exist
- * and are accessible
+ * {@code MissingTopicClassifier} classifies missing source topic exceptions as user error
  */
 public class MissingTopicClassifier implements QueryErrorClassifier {
 

ksqldb-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java

@@ -227,7 +227,8 @@ public PersistentQueryMetadata buildPersistentQuery(
 
     final QueryErrorClassifier topicClassifier = new MissingTopicClassifier(applicationId);
     final QueryErrorClassifier authorizationClassifier =
-        new TopicAuthorizationClassifier(applicationId);
+        new TopicAuthorizationClassifier(applicationId)
+            .and(new GroupAuthorizationClassifier(applicationId));
     final QueryErrorClassifier classifier = buildConfiguredClassifiers(ksqlConfig, applicationId)
         .map(topicClassifier::and)
         .orElse(topicClassifier)

ksqldb-engine/src/main/java/io/confluent/ksql/query/TopicAuthorizationClassifier.java

@@ -18,12 +18,12 @@
 import io.confluent.ksql.query.QueryError.Type;
 import java.util.Objects;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
+import org.apache.kafka.streams.errors.StreamsException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * {@code TopicAuthorizationClassifier} classifies errors by querying the broker
- * to check that all used topics are being accessible.
+ * {@code TopicAuthorizationClassifier} classifies topic ACL exceptions as user error
  */
 public class TopicAuthorizationClassifier implements QueryErrorClassifier {
 
@@ -41,7 +41,7 @@ public Type classify(final Throwable e) {
 
     if (type == Type.USER) {
       LOG.info(
-          "Classified error as USER error based on missing access rights. Query ID: {} Exception: {}",
+          "Classified error as USER error based on missing topic access rights. Query ID: {} Exception: {}",
           queryId,
           e);
     }

ksqldb-engine/src/test/java/io/confluent/ksql/integration/SecureIntegrationTest.java

@@ -34,6 +34,8 @@
 import static org.apache.kafka.common.resource.ResourceType.CLUSTER;
 import static org.apache.kafka.common.resource.ResourceType.GROUP;
 import static org.apache.kafka.common.resource.ResourceType.TOPIC;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 
@@ -44,6 +46,8 @@
 import io.confluent.ksql.engine.KsqlEngineTestUtil;
 import io.confluent.ksql.function.InternalFunctionRegistry;
 import io.confluent.ksql.logging.processing.ProcessingLogContext;
+import io.confluent.ksql.query.QueryError;
+import io.confluent.ksql.query.QueryError.Type;
 import io.confluent.ksql.query.QueryId;
 import io.confluent.ksql.query.id.SequentialQueryIdGenerator;
 import io.confluent.ksql.services.DisabledKsqlClient;
@@ -60,6 +64,8 @@
 import io.confluent.ksql.util.OrderDataProvider;
 import io.confluent.ksql.util.PersistentQueryMetadata;
 import io.confluent.ksql.util.QueryMetadata;
+import java.time.Duration;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -74,8 +80,11 @@
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.errors.GroupAuthorizationException;
+import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.streams.errors.MissingSourceTopicException;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.ClassRule;
@@ -243,6 +252,148 @@ public void shouldWorkWithMinimalPrefixedAcls() {
     assertCanAccessClusterConfig(prefix);
   }
 
+  @Test
+  public void shouldClassifyMissingSourceTopicExceptionAsUserError() {
+    // Given:
+    final String serviceId = "my-service-id_";  // Defaults to "default_"
+    final String prefix = "_confluent-ksql-" + serviceId;
+
+    final Map<String, Object> ksqlConfig = getKsqlConfig(NORMAL_USER);
+    ksqlConfig.put(KSQL_SERVICE_ID_CONFIG, serviceId);
+
+    givenAllowAcl(NORMAL_USER,
+        resource(CLUSTER, "kafka-cluster"),
+        ops(DESCRIBE_CONFIGS));
+
+    givenAllowAcl(NORMAL_USER,
+        resource(TOPIC, INPUT_TOPIC),
+        ops(READ));
+
+    givenAllowAcl(NORMAL_USER,
+        resource(TOPIC, outputTopic),
+        ops(CREATE /* as the topic doesn't exist yet*/, WRITE));
+
+    givenAllowAcl(NORMAL_USER,
+        prefixedResource(TOPIC, prefix),
+        ops(ALL));
+
+    givenAllowAcl(NORMAL_USER,
+        prefixedResource(GROUP, prefix),
+        ops(ALL));
+
+    givenTestSetupWithConfig(ksqlConfig);
+
+    topicClient.deleteTopics(Collections.singleton(INPUT_TOPIC));
+
+    // Then:
+    assertQueryFailsWithUserError(
+        String.format(
+            "CREATE STREAM %s WITH (VALUE_FORMAT='AVRO') AS SELECT * FROM %s;",
+            outputTopic,
+            INPUT_STREAM
+        ),
+        String.format(
+            "%s: One or more source topics were missing during rebalance",
+            MissingSourceTopicException.class.getName()
+        )
+    );
+  }
+
+  @Test
+  public void shouldClassifyTopicAuthorizationExceptionAsUserError() {
+    // Given:
+    final String serviceId = "my-service-id_";  // Defaults to "default_"
+    final String prefix = "_confluent-ksql-" + serviceId;
+
+    final Map<String, Object> ksqlConfig = getKsqlConfig(NORMAL_USER);
+    ksqlConfig.put(KSQL_SERVICE_ID_CONFIG, serviceId);
+
+    givenAllowAcl(NORMAL_USER,
+        resource(CLUSTER, "kafka-cluster"),
+        ops(DESCRIBE_CONFIGS));
+
+    givenAllowAcl(NORMAL_USER,
+        resource(TOPIC, INPUT_TOPIC),
+        ops(READ));
+
+    givenAllowAcl(NORMAL_USER,
+        prefixedResource(TOPIC, prefix),
+        ops(ALL));
+
+    givenAllowAcl(NORMAL_USER,
+        resource(TOPIC, outputTopic),
+        ops(CREATE /* as the topic doesn't exist yet*/, WRITE));
+
+    givenAllowAcl(NORMAL_USER,
+        prefixedResource(GROUP, prefix),
+        ops(ALL));
+
+    givenTestSetupWithConfig(ksqlConfig);
+
+    TEST_HARNESS.getKafkaCluster().addUserAcl(
+        NORMAL_USER.username,
+        AclPermissionType.DENY,
+        resource(TOPIC, INPUT_TOPIC),
+        ops(READ)
+    );
+
+    // Then:
+    assertQueryFailsWithUserError(
+        String.format(
+            "CREATE STREAM %s WITH (VALUE_FORMAT='AVRO') AS SELECT * FROM %s;",
+            outputTopic,
+            INPUT_STREAM
+        ),
+        String.format(
+            "%s: Not authorized to access topics: [%s]",
+            TopicAuthorizationException.class.getName(),
+            INPUT_TOPIC
+        )
+    );
+  }
+
+  @Test
+  public void shouldClassifyGroupAuthorizationExceptionAsUserError() {
+    // Given:
+    final String serviceId = "my-service-id_";  // Defaults to "default_"
+    final String prefix = "_confluent-ksql-" + serviceId;
+
+    final Map<String, Object> ksqlConfig = getKsqlConfig(NORMAL_USER);
+    ksqlConfig.put(KSQL_SERVICE_ID_CONFIG, serviceId);
+
+    givenAllowAcl(NORMAL_USER,
+        resource(CLUSTER, "kafka-cluster"),
+        ops(DESCRIBE_CONFIGS));
+
+    givenAllowAcl(NORMAL_USER,
+        resource(TOPIC, INPUT_TOPIC),
+        ops(READ));
+
+    givenAllowAcl(NORMAL_USER,
+        resource(TOPIC, outputTopic),
+        ops(CREATE /* as the topic doesn't exist yet*/, WRITE));
+
+    givenAllowAcl(NORMAL_USER,
+        prefixedResource(TOPIC, prefix),
+        ops(ALL));
+
+    givenTestSetupWithConfig(ksqlConfig);
+
+    // Then:
+    assertQueryFailsWithUserError(
+        String.format(
+            "CREATE STREAM %s WITH (VALUE_FORMAT='AVRO') AS SELECT * FROM %s;",
+            outputTopic,
+            INPUT_STREAM
+        ),
+        String.format(
+            "%s: Not authorized to access group: %squery_",
+            GroupAuthorizationException.class.getName(),
+            prefix
+        ) + "%s"
+    );
+  }
+
   // Requires correctly configured schema-registry running
   //@Test
   @SuppressWarnings("unused")
@@ -321,6 +472,29 @@ private void assertCanRunKsqlQuery(
     TEST_HARNESS.verifyAvailableRecords(outputTopic, greaterThan(0));
   }
 
+  private void assertQueryFailsWithUserError(
+      final String query,
+      final String errorMsg
+  ) {
+    final QueryMetadata queryMetadata = KsqlEngineTestUtil
+        .execute(serviceContext, ksqlEngine, query, ksqlConfig, Collections.emptyMap()).get(0);
+
+    queryMetadata.start();
+    assertThatEventually(
+        "",
+        () -> queryMetadata.getQueryErrors().size() > 0,
+        is(true)
+    );
+
+    for (final QueryError error : queryMetadata.getQueryErrors()) {
+        assertThat(error.getType(), is(Type.USER));
+        assertThat(
+            error.getErrorMessage().split("\n")[0],
+            is(String.format(errorMsg, queryMetadata.getQueryId()))
+        );
+    }
+  }
+
   private static Map<String, Object> getBaseKsqlConfig() {
     final Map<String, Object> configs = new HashMap<>(KsqlConfigTestUtil.baseTestConfig());
     configs.put(
@@ -382,6 +556,6 @@ private void executePersistentQuery(final String queryString,
         .execute(serviceContext, ksqlEngine, query, ksqlConfig, Collections.emptyMap()).get(0);
 
     queryMetadata.start();
-    queryId = ((PersistentQueryMetadata) queryMetadata).getQueryId();
+    queryId = queryMetadata.getQueryId();
   }
 }

ksqldb-engine/src/test/java/io/confluent/ksql/query/GroupAuthorizationClassifierTest.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2021 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"; you may not use
+ * this file except in compliance with the License. You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.query;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import io.confluent.ksql.query.QueryError.Type;
+import org.apache.kafka.common.errors.GroupAuthorizationException;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class GroupAuthorizationClassifierTest {
+
+  @Test
+  public void shouldClassifyTopicAuthorizationExceptionAsUserError() {
+    // Given:
+    final Exception e = new StreamsException(new GroupAuthorizationException("foo"));
+
+    // When:
+    final Type type = new GroupAuthorizationClassifier("").classify(e);
+
+    // Then:
+    assertThat(type, is(Type.USER));
+  }
+
+  @Test
+  public void shouldClassifyNoGroupAuthorizationExceptionAsUnknownError() {
+    // Given:
+    final Exception e = new Exception("foo");
+
+    // When:
+    final Type type = new GroupAuthorizationClassifier("").classify(e);
+
+    // Then:
+    assertThat(type, is(Type.UNKNOWN));
+  }
+
+}

ksqldb-engine/src/test/java/io/confluent/ksql/query/TopicAuthorizationClassifierTest.java

@@ -21,6 +21,7 @@
 import io.confluent.ksql.query.QueryError.Type;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.streams.errors.MissingSourceTopicException;
+import org.apache.kafka.streams.errors.StreamsException;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.junit.MockitoJUnitRunner;
@@ -31,7 +32,7 @@ public class TopicAuthorizationClassifierTest {
   @Test
   public void shouldClassifyTopicAuthorizationExceptionAsUserError() {
     // Given:
-    final Exception e = new TopicAuthorizationException("foo");
+    final Exception e = new StreamsException(new TopicAuthorizationException("foo"));
 
     // When:
     final Type type = new TopicAuthorizationClassifier("").classify(e);