Merge branch '8.0.x' into master by rayokota

Author: semaphore-agent-production[bot]

bin/ksql-run-class

@@ -53,6 +53,15 @@ if [ -z "$KSQL_LOG4J_OPTS" ]; then
   elif [ -e "/etc/ksqldb/log4j2.yaml" ]; then # Normal install layout
     KSQL_LOG4J_OPTS="-Dlog4j2.configurationFile=file:/etc/ksqldb/log4j2.yaml"
   fi
+else
+  if echo "$KSQL_LOG4J_OPTS" | grep -E "log4j\.[^[:space:]]+(\.properties|\.xml)$"; then
+      # Enable Log4j 1.x configuration compatibility mode for Log4j 2
+      export LOG4J_COMPATIBILITY=true
+      echo "DEPRECATED: A Log4j 1.x configuration file has been detected, which is no longer recommended." >&2
+      echo "To use a Log4j 2.x configuration, please see https://logging.apache.org/log4j/2.x/migrate-from-log4j1.html#Log4j2ConfigurationFormat for details about Log4j configuration file migration." >&2
+      echo "Since ksql processing log functionality uses kafka log4j appender, which was deprecated from 8.0.x, it will not be supported when using log4j1." >&2
+      echo "To continue using this functionality please switch to log4j2." >&2
+  fi
 fi
 
  KSQL_LOG4J_OPTS="-Dksql.log.dir=$LOG_DIR ${KSQL_LOG4J_OPTS}"

ksqldb-engine/pom.xml

@@ -28,6 +28,11 @@
     <artifactId>ksqldb-engine</artifactId>
 
     <dependencies>
+        <dependency>
+            <groupId>com.jayway.jsonpath</groupId>
+            <artifactId>json-path</artifactId>
+            <version>2.9.0</version>
+        </dependency>
         <dependency>
             <groupId>io.confluent.ksql</groupId>
             <artifactId>ksqldb-common</artifactId>

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

@@ -78,7 +78,7 @@ public Optional<ParsedSchema> parseSchema(
         final String schemaType,
         final String schemaString,
         final List<SchemaReference> references) {
-      throw new UnsupportedOperationException();
+      return sandboxCacheClient.parseSchema(schemaType, schemaString, references);
     }
 
     @Override
@@ -286,6 +286,21 @@ public int getId(final String subject, final ParsedSchema parsedSchema)
       }
     }
 
+    @Override
+    public RegisterSchemaResponse getIdWithResponse(
+        final String subject, final ParsedSchema schema, final boolean normalize)
+        throws IOException, RestClientException {
+      try {
+        return srClient.getIdWithResponse(subject, schema,  normalize);
+      } catch (final RestClientException e) {
+        // if we don't find the schema in SR, we try to get it from the sandbox cache
+        if (e.getStatus() == HttpStatus.SC_NOT_FOUND) {
+          return sandboxCacheClient.getIdWithResponse(subject, schema, normalize);
+        }
+        throw e;
+      }
+    }
+
     @Override
     public void reset() {
       throw new UnsupportedOperationException();

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

@@ -20,6 +20,7 @@
 import static org.hamcrest.Matchers.notNullValue;
 import static org.junit.Assert.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
@@ -31,13 +32,18 @@
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import io.confluent.kafka.schemaregistry.client.SchemaMetadata;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
+import io.confluent.kafka.schemaregistry.client.rest.entities.Metadata;
+import io.confluent.kafka.schemaregistry.client.rest.entities.RuleSet;
 import io.confluent.kafka.schemaregistry.client.rest.entities.requests.RegisterSchemaResponse;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
 import io.confluent.ksql.test.util.TestMethods;
 import io.confluent.ksql.test.util.TestMethods.TestCase;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 
+import java.util.Optional;
 import org.apache.avro.Schema;
 import org.apache.hc.core5.http.HttpStatus;
 import org.junit.Before;
@@ -78,8 +84,12 @@ public static Collection<TestCase<SchemaRegistryClient>> getMethodsToTest() {
           .ignore("getId", String.class, ParsedSchema.class)
           .ignore("getId", String.class, ParsedSchema.class, boolean.class)
           .ignore("getId", String.class, Schema.class)
+          .ignore("getIdWithResponse", String.class, ParsedSchema.class, boolean.class)
           .ignore("getVersion", String.class, ParsedSchema.class)
           .ignore("getSchemaById", int.class)
+          .ignore("parseSchema", io.confluent.kafka.schemaregistry.client.rest.entities.Schema.class)
+          .ignore("parseSchema", String.class, String.class, List.class)
+          .ignore("parseSchema", String.class, String.class, List.class, Metadata.class, RuleSet.class)
           .build();
     }
 
@@ -251,7 +261,52 @@ public void shouldGetIdFromCache() throws Exception {
 
       // Then:
       assertThat(id, is(newId));
+    }
+
+    @Test
+    public void shouldGetIdWithResponse() throws Exception {
+      // Given:
+      when(delegate.getIdWithResponse(anyString(), any(ParsedSchema.class), anyBoolean()))
+          .thenReturn(new RegisterSchemaResponse(123))
+          .thenReturn(new RegisterSchemaResponse(124))
+          .thenReturn(new RegisterSchemaResponse(125));
 
+      // When:
+      final int id = sandboxedClient.getIdWithResponse("some subject", schema, false).getId();
+      final int id1 = sandboxedClient.getIdWithResponse("some subject", parsedSchema, false).getId();
+      final int id2 = sandboxedClient.getIdWithResponse("some subject", parsedSchema, true).getId();
+
+      // Then:
+      assertThat(id, is(123));
+      assertThat(id1, is(124));
+      assertThat(id2, is(125));
+    }
+
+    @Test
+    public void shouldGetIdWithResponseFromCache() throws Exception {
+      // Given:
+      final RestClientException exception = mock(RestClientException.class);
+      when(exception.getStatus()).thenReturn(HttpStatus.SC_NOT_FOUND);
+      when(delegate.getIdWithResponse(anyString(), any(ParsedSchema.class), anyBoolean()))
+          .thenThrow(exception);
+
+      final int newId = sandboxedClient.register("newSubject", parsedSchema);
+
+      // When:
+      final int id = sandboxedClient.getIdWithResponse("newSubject", parsedSchema, false).getId();
+
+      // Then:
+      assertThat(id, is(newId));
+    }
+
+    @Test
+    public void shouldParseSchema() throws Exception {
+      // Given:
+      final Optional<ParsedSchema> schema =
+          sandboxedClient.parseSchema("AVRO", "\"string\"", Collections.emptyList());
+
+      // Then:
+      assertThat(schema.get(), is(new AvroSchema("\"string\"")));
     }
   }
 }

ksqldb-rest-app/pom.xml

@@ -99,7 +99,7 @@
 
         <dependency>
             <groupId>io.confluent</groupId>
-            <artifactId>confluent-log4j-extensions</artifactId>
+            <artifactId>confluent-log4j2-extensions</artifactId>
             <version>${io.confluent.common.version}</version>
         </dependency>
 

licenses/LICENSE-zookeeper-3.8.4.txt

@@ -483,6 +483,12 @@
                 <groupId>io.confluent</groupId>
                 <artifactId>broker-plugins</artifactId>
                 <version>${kafka.version}</version>
+                <exclusions>
+                    <exclusion>
+                        <groupId>ch.qos.reload4j</groupId>
+                        <artifactId>reload4j</artifactId>
+                    </exclusion>
+                </exclusions>
             </dependency>
 
             <!-- End Confluent dependencies -->
@@ -788,10 +794,16 @@
             <groupId>io.confluent</groupId>
             <artifactId>logredactor</artifactId>
         </dependency>
-            <dependency>
-                <groupId>io.confluent</groupId>
-                <artifactId>broker-plugins</artifactId>
-            </dependency>
+        <dependency>
+            <groupId>io.confluent</groupId>
+            <artifactId>broker-plugins</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>ch.qos.reload4j</groupId>
+                    <artifactId>reload4j</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 
     <build>