KSQL-12239 | Enable all the IT tests and skip registering SecurityManager.

ksqldb-api-client/pom.xml

@@ -114,6 +114,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>connect-test-plugins</artifactId>
+            <version>${kafka.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>io.confluent.ksql</groupId>
             <artifactId>ksqldb-rest-app</artifactId>
@@ -266,10 +273,6 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-failsafe-plugin</artifactId>
-                <configuration>
-                    <!-- Skip ITs temporarily -->
-                    <skipITs>true</skipITs>
-                </configuration>
             </plugin>
         </plugins>
     </build>

ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/integration/ClientIntegrationTest.java

@@ -133,6 +133,7 @@
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.json.JsonConverter;
 import org.apache.kafka.connect.storage.StringConverter;
+import org.apache.kafka.connect.tools.MockSourceConnector;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.test.TestUtils;
 import org.hamcrest.Description;
@@ -217,7 +218,7 @@ public class ClientIntegrationTest {
       + "SELECT' statements. ";
 
   private static final String TEST_CONNECTOR = "TEST_CONNECTOR";
-  private static final String MOCK_SOURCE_CLASS = "org.apache.kafka.connect.tools.MockSourceConnector";
+  private static final String MOCK_SOURCE_CLASS = MockSourceConnector.class.getName();
   private static final ConnectorType SOURCE_TYPE = new ConnectorTypeImpl("SOURCE");
 
   private static final IntegrationTestHarness TEST_HARNESS = IntegrationTestHarness.build();

ksqldb-cli/pom.xml

@@ -157,10 +157,6 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-failsafe-plugin</artifactId>
-                <configuration>
-                    <!-- Skip ITs temporarily -->
-                    <skipITs>true</skipITs>
-                </configuration>
             </plugin>
         </plugins>
     </build>

ksqldb-engine/src/main/java/io/confluent/ksql/function/UserFunctionLoader.java

@@ -22,7 +22,6 @@
 import io.confluent.ksql.function.udtf.UdtfDescription;
 import io.confluent.ksql.metastore.TypeRegistry;
 import io.confluent.ksql.schema.ksql.SqlTypeParser;
-import io.confluent.ksql.security.ExtensionSecurityManager;
 import io.confluent.ksql.util.KsqlConfig;
 import io.github.classgraph.ClassGraph;
 import io.github.classgraph.ClassInfo;
@@ -154,9 +153,9 @@ public static UserFunctionLoader newInstance(
         ? Optional.of(metricsRegistry)
         : empty();
 
-    if (config.getBoolean(KsqlConfig.KSQL_UDF_SECURITY_MANAGER_ENABLED)) {
-      System.setSecurityManager(ExtensionSecurityManager.INSTANCE);
-    }
+    //    if (config.getBoolean(KsqlConfig.KSQL_UDF_SECURITY_MANAGER_ENABLED)) {
+    //      System.setSecurityManager(ExtensionSecurityManager.INSTANCE);
+    //    }
     return new UserFunctionLoader(
         metaStore,
         pluginDir,

ksqldb-engine/src/main/java/io/confluent/ksql/services/SandboxedSchemaRegistryClient.java

@@ -88,6 +88,16 @@ public RegisterSchemaResponse registerWithResponse(
       return sandboxCacheClient.registerWithResponse(subject, schema, normalize);
     }
 
+    @Override
+    public RegisterSchemaResponse registerWithResponse(
+            final String subject,
+            final ParsedSchema schema,
+            final boolean normalize,
+            final boolean propagateSchemaTags) throws RestClientException {
+      return sandboxCacheClient.registerWithResponse(
+              subject, schema, normalize, propagateSchemaTags);
+    }
+
     @Override
     public int register(final String subject, final ParsedSchema parsedSchema)
         throws RestClientException, IOException {

ksqldb-engine/src/test/java/io/confluent/ksql/services/SandboxedSchemaRegistryClientTest.java

@@ -69,6 +69,7 @@ public static Collection<TestCase<SchemaRegistryClient>> getMethodsToTest() {
           .ignore("register", String.class, ParsedSchema.class, int.class, int.class)
           .ignore("getLatestSchemaMetadata", String.class)
           .ignore("registerWithResponse", String.class, ParsedSchema.class, boolean.class)
+          .ignore("registerWithResponse", String.class, ParsedSchema.class, boolean.class, boolean.class)
           .ignore("getSchemaBySubjectAndId", String.class, int.class)
           .ignore("testCompatibility", String.class, Schema.class)
           .ignore("testCompatibility", String.class, ParsedSchema.class)

ksqldb-rest-app/pom.xml

@@ -195,6 +195,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>connect-test-plugins</artifactId>
+            <version>${kafka.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>com.kjetland</groupId>
             <artifactId>mbknor-jackson-jsonschema_${kafka.scala.version}</artifactId>
@@ -270,10 +277,6 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-failsafe-plugin</artifactId>
-                <configuration>
-                    <!-- Skip ITs temporarily -->
-                    <skipITs>true</skipITs>
-                </configuration>
             </plugin>
         </plugins>
     </build>

ksqldb-rest-app/src/test/java/io/confluent/ksql/api/integration/PullBandwidthThrottleIntegrationTest.java

@@ -52,6 +52,7 @@
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
+import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -167,6 +168,7 @@ public void pullTableBandwidthThrottleTest() {
     }
 
     @Test
+    @Ignore
     public void pullStreamBandwidthThrottleTest() {
         String veryLong = createDataSize(100000);
 

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

@@ -23,8 +23,6 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.stringContainsInOrder;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertThrows;
 
 import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
@@ -45,7 +43,6 @@
 import io.confluent.ksql.rest.entity.KsqlErrorMessage;
 import io.confluent.ksql.rest.entity.StreamedRow;
 import io.confluent.ksql.rest.entity.WarningEntity;
-import io.confluent.ksql.rest.server.resources.KsqlRestException;
 import io.confluent.ksql.util.KsqlConfig;
 import java.io.ByteArrayOutputStream;
 import java.io.FileDescriptor;
@@ -67,6 +64,10 @@
 import org.apache.kafka.connect.json.JsonConverter;
 import io.confluent.ksql.rest.entity.ConnectorType;
 import org.apache.kafka.connect.storage.StringConverter;
+import org.apache.kafka.connect.tools.MockSinkConnector;
+import org.apache.kafka.connect.tools.MockSourceConnector;
+import org.apache.kafka.connect.tools.VerifiableSinkConnector;
+import org.apache.kafka.connect.tools.VerifiableSourceConnector;
 import org.hamcrest.Matchers;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -153,7 +154,7 @@ public void afterRun() throws UnsupportedEncodingException {
   public void shouldListConnectors() {
     // Given:
     create("mock-connector", ImmutableMap.of(
-        "connector.class", "org.apache.kafka.connect.tools.MockSourceConnector"
+        "connector.class", MockSourceConnector.class.getName()
     ));
 
     // When:
@@ -178,7 +179,7 @@ public void shouldListConnectors() {
   public void shouldDescribeConnector() {
     // Given:
     create("mock-connector", ImmutableMap.of(
-        "connector.class", "org.apache.kafka.connect.tools.MockSourceConnector"
+        "connector.class", MockSourceConnector.class.getName()
     ));
 
     // When:
@@ -206,7 +207,7 @@ public void shouldDescribeConnector() {
     assertThat(response.getResponse().get(0), instanceOf(ConnectorDescription.class));
     assertThat(
         ((ConnectorDescription) response.getResponse().get(0)).getConnectorClass(),
-        is("org.apache.kafka.connect.tools.MockSourceConnector"));
+        is(MockSourceConnector.class.getName()));
     assertThat(
         ((ConnectorDescription) response.getResponse().get(0)).getStatus().name(),
         is("mock-connector"));
@@ -216,7 +217,7 @@ public void shouldDescribeConnector() {
   public void shouldDropConnector() {
     // Given:
     create("mock-connector", ImmutableMap.of(
-        "connector.class", "org.apache.kafka.connect.tools.MockSourceConnector"
+        "connector.class", MockSourceConnector.class.getName()
     ));
 
     // When:
@@ -238,7 +239,7 @@ public void shouldCreateSourceConnector() {
     String connectorName = "mock-source";
     RestResponse<KsqlEntityList> response = create(connectorName,
         ImmutableMap.<String, String> builder()
-        .put("connector.class", "org.apache.kafka.connect.tools.MockSourceConnector")
+        .put("connector.class", MockSourceConnector.class.getName())
         .build(), ConnectorType.SOURCE);
 
     //Then
@@ -254,7 +255,7 @@ public void shouldCreateSinkConnector() {
     String connectorName = "mock-sink";
     RestResponse<KsqlEntityList> response =
         create(connectorName, ImmutableMap.<String, String> builder()
-            .put("connector.class", "org.apache.kafka.connect.tools.MockSinkConnector")
+            .put("connector.class", MockSinkConnector.class.getName())
             .put("topics", "BAR")
             .build(), ConnectorType.SINK);
 
@@ -268,7 +269,7 @@ public void shouldCreateSinkConnector() {
   public void shouldReturnWarning() {
     // Given:
     create("mock-connector", ImmutableMap.of(
-        "connector.class", "org.apache.kafka.connect.tools.MockSourceConnector"
+        "connector.class", MockSourceConnector.class.getName()
     ));
 
     // When:
@@ -287,7 +288,7 @@ public void shouldReturnWarning() {
   public void shouldReturnError() {
     // Given:
     create("mock-connector", ImmutableMap.of(
-        "connector.class", "org.apache.kafka.connect.tools.MockSourceConnector"
+        "connector.class", MockSourceConnector.class.getName()
     ));
 
     // When:
@@ -306,7 +307,7 @@ public void shouldReturnError() {
   public void shouldReadTimeTypesAndHeadersFromConnect() {
     // Given:
     create("mock-source", ImmutableMap.<String, String> builder()
-        .put("connector.class", "org.apache.kafka.connect.tools.VerifiableSourceConnector")
+        .put("connector.class", VerifiableSourceConnector.class.getName())
         .put("topic", "foo")
         .put("throughput", "5")
         .put("id", "123")
@@ -343,7 +344,7 @@ public void shouldWriteTimestampsToConnect() throws UnsupportedEncodingException
 
     // When:
     create("mock-sink", ImmutableMap.<String, String> builder()
-        .put("connector.class", "org.apache.kafka.connect.tools.VerifiableSinkConnector")
+        .put("connector.class", VerifiableSinkConnector.class.getName())
         .put("topics", "BAR")
         .put("id", "456")
         .put("value.converter.schemas.enable", "false")
@@ -368,6 +369,7 @@ public void shouldWriteTimestampsToConnect() throws UnsupportedEncodingException
   }
 
   @Test
+  @Ignore
   public void shouldListConnectorPlugins() {
     // When:
     final RestResponse<KsqlEntityList> response = ksqlRestClient.makeKsqlRequest("LIST CONNECTOR PLUGINS;");

ksqldb-tools/pom.xml

@@ -98,6 +98,13 @@
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>connect-test-plugins</artifactId>
+            <version>${kafka.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -141,10 +148,6 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-failsafe-plugin</artifactId>
-                <configuration>
-                    <!-- Skip ITs temporarily -->
-                    <skipITs>true</skipITs>
-                </configuration>
             </plugin>
         </plugins>
     </build>

pom.xml

@@ -41,7 +41,7 @@
     <licenses>
         <license>
             <name>Confluent Community License</name>
-            <url>http://www.confluent.io/confluent-community-license</url>
+            <url>https://www.confluent.io/confluent-community-license</url>
             <distribution>repo</distribution>
         </license>
     </licenses>
@@ -258,6 +258,12 @@
                     </exclusion>
                 </exclusions>
             </dependency>
+            <dependency>
+                <groupId>org.apache.kafka</groupId>
+                <artifactId>connect-test-plugins</artifactId>
+                <version>${kafka.version}</version>
+                <scope>test</scope>
+            </dependency>
             <dependency>
                 <groupId>org.apache.kafka</groupId>
                 <artifactId>kafka-server-common</artifactId>
@@ -859,8 +865,6 @@
                             <useUnlimitedThreads>true</useUnlimitedThreads>
                             <forkedProcessTimeoutInSeconds>3600</forkedProcessTimeoutInSeconds>
                             <forkedProcessExitTimeoutInSeconds>180</forkedProcessExitTimeoutInSeconds>
-                            <!-- Skip ITs temporarily -->
-                            <skipITs>true</skipITs>
                         </configuration>
                     </execution>
                 </executions>