AK to CCS 2.5 Merge

README.md

@@ -92,6 +92,10 @@ This is for `core`, `examples` and `clients`
     ./gradlew core:jar
     ./gradlew core:test
 
+Streams has multiple sub-projects, but you can run all the tests:
+
+    ./gradlew :streams:testAll
+
 ### Listing all gradle tasks ###
     ./gradlew tasks
 

build.gradle

@@ -329,6 +329,7 @@ subprojects {
 
   test {
     maxParallelForks = userMaxForks ?: Runtime.runtime.availableProcessors()
+    ignoreFailures = userIgnoreFailures
 
     minHeapSize = "256m"
     maxHeapSize = "2048m"
@@ -355,6 +356,10 @@ subprojects {
     }
     logTestStdout.rehydrate(delegate, owner, this)()
 
+    // The suites are for running sets of tests in IDEs.
+    // Gradle will run each test class, so we exclude the suites to avoid redundantly running the tests twice.
+    exclude '**/*Suite.class'
+
     useJUnit {
       includeCategories 'org.apache.kafka.test.IntegrationTest'
     }
@@ -374,6 +379,10 @@ subprojects {
       exceptionFormat = testExceptionFormat
     }
     logTestStdout.rehydrate(delegate, owner, this)()
+
+    // The suites are for running sets of tests in IDEs.
+    // Gradle will run each test class, so we exclude the suites to avoid redundantly running the tests twice.
+    exclude '**/*Suite.class'
 
     if (it.project.name != 'generator') {
       useJUnit {
@@ -499,8 +508,8 @@ subprojects {
       scalaCompileOptions.additionalParameters += inlineFrom
     }
 
-  // these options are valid for Scala versions < 2.13 only
-  // Scala 2.13 removes them, see https://github.com/scala/scala/pull/6502 and https://github.com/scala/scala/pull/5969
+    // these options are valid for Scala versions < 2.13 only
+    // Scala 2.13 removes them, see https://github.com/scala/scala/pull/6502 and https://github.com/scala/scala/pull/5969
     if (versions.baseScala in ['2.11','2.12']) {
       scalaCompileOptions.additionalParameters += [
         "-Xlint:by-name-right-associative",
@@ -1317,11 +1326,26 @@ project(':streams') {
     standardOutput = new File(generatedDocsDir, "streams_config.html").newOutputStream()
   }
 
-  test {
-    // The suites are for running sets of tests in IDEs.
-    // Gradle will run each test class, so we exclude the suites to avoid redundantly running the tests twice.
-    exclude '**/*Suite.class'
-  }
+  task testAll(
+    dependsOn: [
+            ':streams:test',
+            ':streams:test-utils:test',
+            ':streams:streams-scala:test',
+            ':streams:upgrade-system-tests-0100:test',
+            ':streams:upgrade-system-tests-0101:test',
+            ':streams:upgrade-system-tests-0102:test',
+            ':streams:upgrade-system-tests-0110:test',
+            ':streams:upgrade-system-tests-10:test',
+            ':streams:upgrade-system-tests-11:test',
+            ':streams:upgrade-system-tests-20:test',
+            ':streams:upgrade-system-tests-21:test',
+            ':streams:upgrade-system-tests-22:test',
+            ':streams:upgrade-system-tests-23:test',
+            ':streams:upgrade-system-tests-24:test',
+            ':streams:upgrade-system-tests-25:test',
+            ':streams:examples:test'
+    ]
+  )
 }
 
 project(':streams:streams-scala') {

clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java

@@ -107,7 +107,7 @@ public void testResolveDnsLookup() throws UnknownHostException {
 
     @Test
     public void testResolveDnsLookupAllIps() throws UnknownHostException {
-        assertEquals(2, ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.USE_ALL_DNS_IPS).size());
+        assertTrue(ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1);
     }
 
     private List<InetSocketAddress> checkWithoutLookup(String... url) {

clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java

@@ -256,7 +256,7 @@ public void testSingleIPWithUseAll() throws UnknownHostException {
 
     @Test
     public void testMultipleIPsWithDefault() throws UnknownHostException {
-        assertEquals(2, ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size());
+        assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1);
 
         connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.DEFAULT);
         InetAddress currAddress = connectionStates.currentAddress(nodeId1);
@@ -266,22 +266,21 @@ public void testMultipleIPsWithDefault() throws UnknownHostException {
 
     @Test
     public void testMultipleIPsWithUseAll() throws UnknownHostException {
-        assertEquals(2, ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size());
+        assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1);
 
         connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS);
         InetAddress addr1 = connectionStates.currentAddress(nodeId1);
         connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS);
         InetAddress addr2 = connectionStates.currentAddress(nodeId1);
         assertNotSame(addr1, addr2);
-
         connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS);
         InetAddress addr3 = connectionStates.currentAddress(nodeId1);
-        assertSame(addr1, addr3);
+        assertNotSame(addr1, addr3);
     }
 
     @Test
     public void testHostResolveChange() throws UnknownHostException, ReflectiveOperationException {
-        assertEquals(2, ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size());
+        assertTrue(ClientUtils.resolve(hostTwoIps, ClientDnsLookup.USE_ALL_DNS_IPS).size() > 1);
 
         connectionStates.connecting(nodeId1, time.milliseconds(), hostTwoIps, ClientDnsLookup.DEFAULT);
         InetAddress addr1 = connectionStates.currentAddress(nodeId1);

connect/api/src/main/java/org/apache/kafka/connect/data/Values.java

@@ -943,8 +943,14 @@ protected static SchemaAndValue parse(Parser parser, boolean embedded) throws No
                 } catch (ArithmeticException e) {
                     // continue
                 }
+                float fValue = decimal.floatValue();
+                if (fValue != Float.NEGATIVE_INFINITY && fValue != Float.POSITIVE_INFINITY
+                    && decimal.scale() != 0) {
+                    return new SchemaAndValue(Schema.FLOAT32_SCHEMA, fValue);
+                }
                 double dValue = decimal.doubleValue();
-                if (dValue != Double.NEGATIVE_INFINITY && dValue != Double.POSITIVE_INFINITY) {
+                if (dValue != Double.NEGATIVE_INFINITY && dValue != Double.POSITIVE_INFINITY
+                    && decimal.scale() != 0) {
                     return new SchemaAndValue(Schema.FLOAT64_SCHEMA, dValue);
                 }
                 Schema schema = Decimal.schema(decimal.scale());
@@ -1187,7 +1193,7 @@ private String consumeNextToken() throws NoSuchElementException {
             boolean escaped = false;
             int start = iter.getIndex();
             char c = iter.current();
-            while (c != CharacterIterator.DONE) {
+            while (canConsumeNextToken()) {
                 switch (c) {
                     case '\\':
                         escaped = !escaped;

connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java

@@ -21,6 +21,9 @@
 import org.apache.kafka.connect.errors.DataException;
 import org.junit.Test;
 
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -73,6 +76,18 @@ public class ValuesTest {
         INT_LIST.add(-987654321);
     }
 
+    @Test(timeout = 5000)
+    public void shouldNotEncounterInfiniteLoop() {
+        // This byte sequence gets parsed as CharacterIterator.DONE and can cause issues if
+        // comparisons to that character are done to check if the end of a string has been reached.
+        // For more information, see https://issues.apache.org/jira/browse/KAFKA-10574
+        byte[] bytes = new byte[] {-17, -65,  -65};
+        String str = new String(bytes, StandardCharsets.UTF_8);
+        SchemaAndValue schemaAndValue = Values.parseString(str);
+        assertEquals(Type.STRING, schemaAndValue.schema().type());
+        assertEquals(str, schemaAndValue.value());
+    }
+
     @Test
     public void shouldNotParseUnquotedEmbeddedMapKeysAsStrings() {
         SchemaAndValue schemaAndValue = Values.parseString("{foo: 3}");
@@ -726,6 +741,132 @@ public void shouldConvertTimestampValues() {
     public void canConsume() {
     }
 
+    @Test
+    public void shouldParseBigIntegerAsDecimalWithZeroScale() {
+        BigInteger value = BigInteger.valueOf(Long.MAX_VALUE).add(new BigInteger("1"));
+        SchemaAndValue schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Decimal.schema(0), schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof BigDecimal);
+        assertEquals(value, ((BigDecimal) schemaAndValue.value()).unscaledValue());
+        value = BigInteger.valueOf(Long.MIN_VALUE).subtract(new BigInteger("1"));
+        schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Decimal.schema(0), schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof BigDecimal);
+        assertEquals(value, ((BigDecimal) schemaAndValue.value()).unscaledValue());
+    }
+
+    @Test
+    public void shouldParseByteAsInt8() {
+        Byte value = Byte.MAX_VALUE;
+        SchemaAndValue schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.INT8_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Byte);
+        assertEquals(value.byteValue(), ((Byte) schemaAndValue.value()).byteValue());
+        value = Byte.MIN_VALUE;
+        schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.INT8_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Byte);
+        assertEquals(value.byteValue(), ((Byte) schemaAndValue.value()).byteValue());
+    }
+
+    @Test
+    public void shouldParseShortAsInt16() {
+        Short value = Short.MAX_VALUE;
+        SchemaAndValue schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.INT16_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Short);
+        assertEquals(value.shortValue(), ((Short) schemaAndValue.value()).shortValue());
+        value = Short.MIN_VALUE;
+        schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.INT16_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Short);
+        assertEquals(value.shortValue(), ((Short) schemaAndValue.value()).shortValue());
+    }
+
+    @Test
+    public void shouldParseIntegerAsInt32() {
+        Integer value = Integer.MAX_VALUE;
+        SchemaAndValue schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.INT32_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Integer);
+        assertEquals(value.intValue(), ((Integer) schemaAndValue.value()).intValue());
+        value = Integer.MIN_VALUE;
+        schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.INT32_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Integer);
+        assertEquals(value.intValue(), ((Integer) schemaAndValue.value()).intValue());
+    }
+
+    @Test
+    public void shouldParseLongAsInt64() {
+        Long value = Long.MAX_VALUE;
+        SchemaAndValue schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.INT64_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Long);
+        assertEquals(value.longValue(), ((Long) schemaAndValue.value()).longValue());
+        value = Long.MIN_VALUE;
+        schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.INT64_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Long);
+        assertEquals(value.longValue(), ((Long) schemaAndValue.value()).longValue());
+    }
+
+    @Test
+    public void shouldParseFloatAsFloat32() {
+        Float value = Float.MAX_VALUE;
+        SchemaAndValue schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.FLOAT32_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Float);
+        assertEquals(value.floatValue(), ((Float) schemaAndValue.value()).floatValue(), 0);
+        value = -Float.MAX_VALUE;
+        schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.FLOAT32_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Float);
+        assertEquals(value.floatValue(), ((Float) schemaAndValue.value()).floatValue(), 0);
+    }
+
+    @Test
+    public void shouldParseDoubleAsFloat64() {
+        Double value = Double.MAX_VALUE;
+        SchemaAndValue schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.FLOAT64_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Double);
+        assertEquals(value.doubleValue(), ((Double) schemaAndValue.value()).doubleValue(), 0);
+        value = -Double.MAX_VALUE;
+        schemaAndValue = Values.parseString(
+            String.valueOf(value)
+        );
+        assertEquals(Schema.FLOAT64_SCHEMA, schemaAndValue.schema());
+        assertTrue(schemaAndValue.value() instanceof Double);
+        assertEquals(value.doubleValue(), ((Double) schemaAndValue.value()).doubleValue(), 0);
+    }
+
     protected void assertParsed(String input) {
         assertParsed(input, input);
     }

connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java

@@ -729,6 +729,7 @@ private static Object convertToConnect(Schema schema, JsonNode jsonValue) {
         } else {
             switch (jsonValue.getNodeType()) {
                 case NULL:
+                case MISSING:
                     // Special case. With no schema
                     return null;
                 case BOOLEAN:
@@ -751,7 +752,6 @@ private static Object convertToConnect(Schema schema, JsonNode jsonValue) {
                     break;
 
                 case BINARY:
-                case MISSING:
                 case POJO:
                 default:
                     schemaType = null;

connect/json/src/test/java/org/apache/kafka/connect/json/JsonConverterTest.java

@@ -195,6 +195,36 @@ public void nullToConnect() {
         assertEquals(SchemaAndValue.NULL, converted);
     }
 
+    /**
+     * When schemas are disabled, empty data should be decoded to an empty envelope.
+     * This test verifies the case where `schemas.enable` configuration is set to false, and
+     * {@link JsonConverter} converts empty bytes to {@link SchemaAndValue#NULL}.
+     */
+    @Test
+    public void emptyBytesToConnect() {
+        // This characterizes the messages with empty data when Json schemas is disabled
+        Map<String, Boolean> props = Collections.singletonMap("schemas.enable", false);
+        converter.configure(props, true);
+        SchemaAndValue converted = converter.toConnectData(TOPIC, "".getBytes());
+        assertEquals(SchemaAndValue.NULL, converted);
+    }
+
+    /**
+     * When schemas are disabled, fields are mapped to Connect maps.
+     */
+    @Test
+    public void schemalessWithEmptyFieldValueToConnect() {
+        // This characterizes the messages with empty data when Json schemas is disabled
+        Map<String, Boolean> props = Collections.singletonMap("schemas.enable", false);
+        converter.configure(props, true);
+        String input = "{ \"a\": \"\", \"b\": null}";
+        SchemaAndValue converted = converter.toConnectData(TOPIC, input.getBytes());
+        Map<String, String> expected = new HashMap<>();
+        expected.put("a", "");
+        expected.put("b", null);
+        assertEquals(new SchemaAndValue(null, expected), converted);
+    }
+
     @Test
     public void nullSchemaPrimitiveToConnect() {
         SchemaAndValue converted = converter.toConnectData(TOPIC, "{ \"schema\": null, \"payload\": null }".getBytes());