KSQL-12955 | Introduce KsqlResourceExtensions for plugins configured externally.

ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

@@ -55,7 +55,6 @@
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.config.TopicConfig;
-import org.apache.kafka.common.config.internals.ConfluentConfigs;
 import org.apache.kafka.common.utils.AppInfoParser;
 import org.apache.kafka.streams.StreamsConfig;
 import org.slf4j.Logger;
@@ -728,6 +727,13 @@ public enum DeploymentType {
       = "Configure how long the remote host executor will wait for in seconds "
       + "when fetching all remote hosts.";
 
+  public static final String KSQL_ENABLE_FIPS = "enable.fips";
+  public static final String KSQL_ENABLE_FIPS_DEFAULT = "false";
+  public static final String KSQL_ENABLE_FIPS_DOC
+          = "Enable FIPS mode on the server. If FIPS mode is enabled, "
+          + "broker listener security protocols, TLS versions and cipher "
+          + "suites will be validated based on FIPS compliance requirement.";
+
   private enum ConfigGeneration {
     LEGACY,
     CURRENT
@@ -1559,11 +1565,11 @@ private static ConfigDef buildConfigDef(final ConfigGeneration generation) {
         )
         .withClientSslSupport()
         .define(
-            ConfluentConfigs.ENABLE_FIPS_CONFIG,
+            KSQL_ENABLE_FIPS,
             Type.BOOLEAN,
-            ConfluentConfigs.ENABLE_FIPS_DEFAULT,
+            KSQL_ENABLE_FIPS_DEFAULT,
             Importance.LOW,
-            ConfluentConfigs.ENABLE_FIPS_DOC
+            KSQL_ENABLE_FIPS_DOC
         );
 
     for (final CompatibilityBreakingConfigDef compatibilityBreakingConfigDef
@@ -1792,6 +1798,10 @@ public Map<String, Object> getConsumerClientConfigProps() {
     return Collections.unmodifiableMap(map);
   }
 
+  public boolean enableFips() {
+    return getBoolean(KSQL_ENABLE_FIPS);
+  }
+
   public Map<String, Object> addConfluentMetricsContextConfigsKafka(
       final Map<String,Object> props
   ) {
@@ -1940,7 +1950,7 @@ public static Map<String, String> getStringAsMap(
 
   public static Map<String, String> parseStringAsMap(final String key, final String value) {
     try {
-      return value.equals("")
+      return value.isEmpty()
           ? Collections.emptyMap()
           : Splitter.on(",").trimResults().withKeyValueSeparator(":").split(value);
     } catch (final IllegalArgumentException e) {

ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConstants.java

@@ -52,6 +52,9 @@ private KsqlConstants() {
   public static final String KSQL_QUERY_PLAN_TYPE_TAG = "query_plan_type";
   public static final String KSQL_QUERY_ROUTING_TYPE_TAG = "query_routing_type";
 
+  public static final String FIPS_VALIDATOR
+          = "io.confluent.ksql.security.KsqlFipsResourceExtension";
+
   public enum KsqlQueryType {
     PERSISTENT,
     PUSH,

ksqldb-common/src/main/java/io/confluent/ksql/util/MetricsTagsUtil.java

@@ -27,7 +27,7 @@ public static Map<String, String> getMetricsTagsWithQueryId(
       final String queryId,
       final Map<String, String> tags
   ) {
-    if (queryId.equals("")) {
+    if (queryId.isEmpty()) {
       return tags;
     }
     return addMetricTagToMap("query-id", queryId, tags);

ksqldb-engine/src/main/java/io/confluent/ksql/schema/registry/KsqlSchemaRegistryClientFactory.java

@@ -29,7 +29,6 @@
 import java.util.Map;
 import java.util.function.Supplier;
 import javax.net.ssl.SSLContext;
-import org.apache.kafka.common.config.internals.ConfluentConfigs;
 import org.apache.kafka.common.security.auth.SslEngineFactory;
 import org.apache.kafka.common.security.ssl.DefaultSslEngineFactory;
 import org.apache.kafka.common.utils.SecurityUtils;
@@ -97,7 +96,7 @@ public KsqlSchemaRegistryClientFactory(
    * Creates an SslContext configured to be used with the KsqlSchemaRegistryClient.
    */
   public static SSLContext newSslContext(final KsqlConfig config) {
-    if (config.getBoolean(ConfluentConfigs.ENABLE_FIPS_CONFIG)) {
+    if (config.enableFips()) {
       SecurityUtils.addConfiguredSecurityProviders(config.originals());
     }
     final DefaultSslEngineFactory sslFactory = new DefaultSslEngineFactory();
@@ -115,7 +114,7 @@ static void configureSslEngineFactory(
   }
 
   public SchemaRegistryClient get() {
-    if (schemaRegistryUrl.equals("")) {
+    if (schemaRegistryUrl.isEmpty()) {
       return new DefaultSchemaRegistryClient();
     }
 

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/extensions/KsqlResourceContext.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2018 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.rest.extensions;
+
+import io.confluent.ksql.rest.server.KsqlRestConfig;
+import io.confluent.ksql.util.KsqlConfig;
+
+public interface KsqlResourceContext {
+
+  KsqlConfig ksqlConfig();
+
+  KsqlRestConfig ksqlRestConfig();
+}

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/extensions/KsqlResourceContextImpl.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2018 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.rest.extensions;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import io.confluent.ksql.rest.server.KsqlRestConfig;
+import io.confluent.ksql.util.KsqlConfig;
+
+public class KsqlResourceContextImpl implements KsqlResourceContext {
+
+  private final KsqlConfig ksqlConfig;
+  private final KsqlRestConfig ksqlRestConfig;
+
+  @SuppressFBWarnings(value = "EI_EXPOSE_REP2")
+  public KsqlResourceContextImpl(
+          final KsqlConfig ksqlConfig,
+          final KsqlRestConfig restConfig) {
+    this.ksqlConfig = ksqlConfig;
+    this.ksqlRestConfig = restConfig;
+  }
+
+  @Override
+  @SuppressFBWarnings(value = "EI_EXPOSE_REP")
+  public KsqlConfig ksqlConfig() {
+    return ksqlConfig;
+  }
+
+  @Override
+  @SuppressFBWarnings(value = "EI_EXPOSE_REP")
+  public KsqlRestConfig ksqlRestConfig() {
+    return ksqlRestConfig;
+  }
+}

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/extensions/KsqlResourceExtension.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2018 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.rest.extensions;
+
+import io.confluent.ksql.util.KsqlException;
+import java.io.Closeable;
+
+public interface KsqlResourceExtension extends Closeable {
+
+  void register(KsqlResourceContext ksqlResourceContext) throws KsqlException;
+}

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestApplication.java

@@ -60,6 +60,9 @@
 import io.confluent.ksql.rest.entity.KsqlEntityList;
 import io.confluent.ksql.rest.entity.SourceInfo;
 import io.confluent.ksql.rest.entity.StreamsList;
+import io.confluent.ksql.rest.extensions.KsqlResourceContext;
+import io.confluent.ksql.rest.extensions.KsqlResourceContextImpl;
+import io.confluent.ksql.rest.extensions.KsqlResourceExtension;
 import io.confluent.ksql.rest.server.HeartbeatAgent.Builder;
 import io.confluent.ksql.rest.server.computation.Command;
 import io.confluent.ksql.rest.server.computation.CommandRunner;
@@ -125,6 +128,7 @@
 import io.vertx.ext.dropwizard.Match;
 import java.io.Console;
 import java.io.File;
+import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.io.PrintWriter;
 import java.net.MalformedURLException;
@@ -211,6 +215,7 @@ public final class KsqlRestApplication implements Executable {
   private KafkaTopicClient internalTopicClient;
   private final Instant ksqlRestAppStartTime;
   private final KsqlRestApplicationMetrics restApplicationMetrics;
+  private final List<KsqlResourceExtension> ksqlResourceExtensions;
 
   // The startup thread that can be interrupted if necessary during shutdown.  This should only
   // happen if startup hangs.
@@ -279,7 +284,8 @@ public static SourceName getCommandsStreamName() {
     this.vertx = requireNonNull(vertx, "vertx");
     this.denyListPropertyValidator =
         requireNonNull(denyListPropertyValidator, "denyListPropertyValidator");
-
+    this.ksqlResourceExtensions = new KsqlRestConfig(ksqlConfigNoPort.originals())
+              .getKsqlResourceExtensions();
     this.serverInfoResource =
         new ServerInfoResource(serviceContext, ksqlConfigNoPort, commandRunner);
     if (heartbeatAgent.isPresent()) {
@@ -313,6 +319,24 @@ public static SourceName getCommandsStreamName() {
   @Override
   public void startAsync() {
     log.debug("Starting the ksqlDB API server");
+
+    final KsqlResourceContext ksqlResourceContext
+            = new KsqlResourceContextImpl(ksqlConfigNoPort, restConfig);
+
+    boolean isFipsValidatorConfigured = false;
+    for (KsqlResourceExtension resourceExtension : ksqlResourceExtensions) {
+      resourceExtension.register(ksqlResourceContext);
+      if (KsqlConstants.FIPS_VALIDATOR
+              .equals(resourceExtension.getClass().getCanonicalName())) {
+        isFipsValidatorConfigured = true;
+      }
+    }
+    if (ksqlConfigNoPort.enableFips() && !isFipsValidatorConfigured) {
+      throw new KsqlException("Error enabling the FIPS resource extension: `enable.fips` is set"
+              + " to true but the `ksql.resource.extension.class` config is either not configured"
+              + " or does not contain \"" + KsqlConstants.FIPS_VALIDATOR + "\"");
+    }
+
     this.serverMetadataResource = ServerMetadataResource.create(serviceContext, ksqlConfigNoPort);
     final StatementParser statementParser = new StatementParser(ksqlEngine);
     final Optional<KsqlAuthorizationValidator> authorizationValidator =
@@ -509,6 +533,14 @@ public void shutdown() {
       }
     });
 
+    ksqlResourceExtensions.forEach(resourceExtension -> {
+      try {
+        resourceExtension.close();
+      } catch (IOException e) {
+        log.error("Exception while closing resource extension", e);
+      }
+    });
+
     try {
       ksqlEngine.close();
     } catch (final Exception e) {
@@ -612,9 +644,9 @@ public static KsqlRestApplication buildApplication(
       final FunctionRegistry functionRegistry,
       final Instant ksqlRestAppStartTime
   ) {
-
     final Map<String, Object> updatedRestProps = restConfig.getOriginals();
     final KsqlConfig ksqlConfig = new KsqlConfig(restConfig.getKsqlConfigProperties());
+
     final Vertx vertx = Vertx.vertx(
         new VertxOptions()
             .setMaxWorkerExecuteTimeUnit(TimeUnit.MILLISECONDS)

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestConfig.java

@@ -24,6 +24,7 @@
 import io.confluent.ksql.configdef.ConfigValidators;
 import io.confluent.ksql.rest.DefaultErrorMessages;
 import io.confluent.ksql.rest.ErrorMessages;
+import io.confluent.ksql.rest.extensions.KsqlResourceExtension;
 import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.util.KsqlServerException;
@@ -444,6 +445,13 @@ public class KsqlRestConfig extends AbstractConfig {
           KSQL_COMMAND_TOPIC_MIGRATION_MIGRATING
       );
 
+  public static final String KSQL_RESOURCE_EXTENSIONS =
+          "ksql.resource.extension.class";
+  private static final String KSQL_RESOURCE_EXTENSION_DEFAULT = "";
+  private static final String KSQL_RESOURCE_EXTENSION_DOC =
+          "A list of KsqlResourceExtension implementations "
+                  + "to be registered with the ksqlDB server.";
+
   private static final ConfigDef CONFIG_DEF;
 
   static {
@@ -617,6 +625,12 @@ public class KsqlRestConfig extends AbstractConfig {
             ConfigValidators.nullsAllowed(ConfigValidators.validUrl()),
             Importance.HIGH,
             INTERNAL_LISTENER_DOC
+        ).define(
+            KSQL_RESOURCE_EXTENSIONS,
+            Type.LIST,
+            KSQL_RESOURCE_EXTENSION_DEFAULT,
+            Importance.MEDIUM,
+            KSQL_RESOURCE_EXTENSION_DOC
         ).define(
             STREAMED_QUERY_DISCONNECT_CHECK_MS_CONFIG,
             Type.LONG,
@@ -1065,6 +1079,10 @@ public ClientAuth getClientAuthInternal() {
     return getClientAuth(getString(KSQL_INTERNAL_SSL_CLIENT_AUTHENTICATION_CONFIG));
   }
 
+  public List<KsqlResourceExtension> getKsqlResourceExtensions() {
+    return getConfiguredInstances(KSQL_RESOURCE_EXTENSIONS, KsqlResourceExtension.class);
+  }
+
   /**
    * Used to sanitize the first `listener` config.
    *

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlServerMain.java

@@ -194,7 +194,7 @@ static void validateDefaultTopicFormats(final KsqlConfig config) {
 
   @VisibleForTesting
   static void validateFips(final KsqlConfig config, final KsqlRestConfig restConfig) {
-    if (config.getBoolean(ConfluentConfigs.ENABLE_FIPS_CONFIG)) {
+    if (config.enableFips()) {
       final FipsValidator fipsValidator = ConfluentConfigs.buildFipsValidator();
 
       // validate cipher suites and TLS version

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/DistributingExecutor.java

@@ -169,7 +169,7 @@ public StatementExecutorResponse execute(
       final KsqlSecurityContext securityContext
   ) {
     final String commandRunnerWarningString = commandRunnerWarning.get();
-    if (!commandRunnerWarningString.equals("")) {
+    if (!commandRunnerWarningString.isEmpty()) {
       throw new KsqlServerException("Failed to handle Ksql Statement."
           + System.lineSeparator()
           + commandRunnerWarningString);

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

@@ -457,7 +457,7 @@ private static void addCommandRunnerWarning(
       final Supplier<String> commandRunnerWarning
   ) {
     final String commandRunnerIssueString = commandRunnerWarning.get();
-    if (!commandRunnerIssueString.equals("")) {
+    if (!commandRunnerIssueString.isEmpty()) {
       for (final KsqlEntity entity: entityList) {
         entity.updateWarnings(Collections.singletonList(
             new KsqlWarning(commandRunnerIssueString)));

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/KsqlRestApplicationTest.java

@@ -60,6 +60,7 @@
 import io.confluent.ksql.services.KafkaTopicClient;
 import io.confluent.ksql.services.ServiceContext;
 import io.confluent.ksql.util.KsqlConfig;
+import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.version.metrics.VersionCheckerAgent;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
@@ -76,6 +77,7 @@
 import org.apache.kafka.common.metrics.MetricsReporter;
 import org.apache.kafka.streams.StreamsConfig;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
@@ -379,6 +381,17 @@ public void shouldConfigureRocksDBConfigSetter() {
     verify(rocksDBConfigSetterHandler).accept(ksqlConfig);
   }
 
+  @Ignore
+  @Test(expected = KsqlException.class)
+  public void shouldFailIfFipsValidationEnabledButNotConfigured() {
+    // When:
+    when(ksqlConfig.enableFips()).thenReturn(true);
+    app.startKsql(ksqlConfig);
+
+    // Then:
+    // KsqlException
+  }
+
   @Test
   public void shouldConfigureIQWithInterNodeListenerIfSet() {
     // Given: