KAFKA-3705: Added a foreignKeyJoin implementation for KTable.

build.gradle

@@ -1224,6 +1224,12 @@ project(':streams') {
     if( !generatedDocsDir.exists() ) { generatedDocsDir.mkdirs() }
     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'
+  }
 }
 
 project(':streams:streams-scala') {

checkstyle/suppressions.xml

@@ -1,4 +1,4 @@
-<?xml version="1.0"?>
+
 
 <!DOCTYPE suppressions PUBLIC
         "-//Puppy Crawl//DTD Suppressions 1.1//EN"
@@ -67,6 +67,8 @@
 
     <suppress checks="(JavaNCSS|CyclomaticComplexity|MethodLength)"
               files="CoordinatorClient.java"/>
+    <suppress checks="(UnnecessaryParentheses|BooleanExpressionComplexity|CyclomaticComplexity|WhitespaceAfter|LocalVariableName)"
+              files="Murmur3.java"/>
 
     <suppress checks="(NPathComplexity|ClassFanOutComplexity|CyclomaticComplexity|ClassDataAbstractionCoupling|LocalVariableName|MemberName|MethodLength|JavaNCSS)"
             files="clients[\\/]src[\\/](generated|generated-test)[\\/].+.java$"/>
@@ -90,6 +92,9 @@
     <suppress checks="NPathComplexity"
               files="MemoryRecordsTest|MetricsTest"/>
 
+    <suppress checks="(WhitespaceAround|LocalVariableName|ImportControl|AvoidStarImport)"
+              files="Murmur3Test.java"/>
+
     <!-- Connect -->
     <suppress checks="ClassFanOutComplexity"
               files="DistributedHerder(|Test).java"/>
@@ -149,7 +154,7 @@
               files="(TopologyBuilder|KafkaStreams|KStreamImpl|KTableImpl|StreamThread|StreamTask).java"/>
 
     <suppress checks="MethodLength"
-              files="StreamsPartitionAssignor.java"/>
+              files="(KTableImpl|StreamsPartitionAssignor.java)"/>
 
     <suppress checks="ParameterNumber"
               files="StreamTask.java"/>
@@ -229,7 +234,8 @@
 
     <suppress checks="CyclomaticComplexity"
               files="(StreamsResetter|ProducerPerformance|Agent).java"/>
-
+    <suppress checks="BooleanExpressionComplexity"
+              files="StreamsResetter.java"/>
     <suppress checks="NPathComplexity"
               files="(ProducerPerformance|StreamsResetter|Agent).java"/>
     <suppress checks="ImportControl"

clients/src/main/java/org/apache/kafka/common/utils/Bytes.java

@@ -141,7 +141,33 @@ private static String toString(final byte[] b, int off, int len) {
     }
 
     /**
-     * A byte array comparator based on lexicographic ordering.
+     * Increment the underlying byte array by adding 1. Throws an IndexOutOfBoundsException if incrementing would cause
+     * the underlying input byte array to overflow.
+     *
+     * @param input - The byte array to increment
+     * @return A new copy of the incremented byte array.
+     */
+    public static Bytes increment(Bytes input) throws IndexOutOfBoundsException {
+        byte[] inputArr = input.get();
+        byte[] ret = new byte[inputArr.length];
+        int carry = 1;
+        for (int i = inputArr.length - 1; i >= 0; i--) {
+            if (inputArr[i] == (byte) 0xFF && carry == 1) {
+                ret[i] = (byte) 0x00;
+            } else {
+                ret[i] = (byte) (inputArr[i] + carry);
+                carry = 0;
+            }
+        }
+        if (carry == 0) {
+            return wrap(ret);
+        } else {
+            throw new IndexOutOfBoundsException();
+        }
+    }
+
+    /**
+     * A byte array comparator based on lexicograpic ordering.
      */
     public final static ByteArrayComparator BYTES_LEXICO_COMPARATOR = new LexicographicByteArrayComparator();
 

clients/src/test/java/org/apache/kafka/common/utils/BytesTest.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.utils;
+
+import org.junit.Test;
+
+import java.util.Comparator;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertEquals;
+
+public class BytesTest {
+
+    @Test
+    public void testIncrement() {
+        byte[] input = new byte[]{(byte) 0xAB, (byte) 0xCD, (byte) 0xFF};
+        byte[] expected = new byte[]{(byte) 0xAB, (byte) 0xCE, (byte) 0x00};
+        Bytes output = Bytes.increment(Bytes.wrap(input));
+        assertArrayEquals(output.get(), expected);
+    }
+
+    @Test
+    public void testIncrementUpperBoundary() {
+        byte[] input = new byte[]{(byte) 0xFF, (byte) 0xFF, (byte) 0xFF};
+        assertThrows(IndexOutOfBoundsException.class, () -> Bytes.increment(Bytes.wrap(input)));
+    }
+
+    @Test
+    public void testIncrementWithSubmap() {
+        final NavigableMap<Bytes, byte[]> map = new TreeMap<>();
+        Bytes key1 = Bytes.wrap(new byte[]{(byte) 0xAA});
+        byte[] val = new byte[]{(byte) 0x00};
+        map.put(key1, val);
+
+        Bytes key2 = Bytes.wrap(new byte[]{(byte) 0xAA, (byte) 0xAA});
+        map.put(key2, val);
+
+        Bytes key3 = Bytes.wrap(new byte[]{(byte) 0xAA, (byte) 0x00, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF});
+        map.put(key3, val);
+
+        Bytes key4 = Bytes.wrap(new byte[]{(byte) 0xAB, (byte) 0x00});
+        map.put(key4, val);
+
+        Bytes key5 = Bytes.wrap(new byte[]{(byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x01});
+        map.put(key5, val);
+
+        Bytes prefix = key1;
+        Bytes prefixEnd = Bytes.increment(prefix);
+
+        Comparator<? super Bytes> comparator = map.comparator();
+        final int result = comparator == null ? prefix.compareTo(prefixEnd) : comparator.compare(prefix, prefixEnd);
+        NavigableMap<Bytes, byte[]> subMapResults;
+        if (result > 0) {
+            //Prefix increment would cause a wrap-around. Get the submap from toKey to the end of the map
+            subMapResults = map.tailMap(prefix, true);
+        } else {
+            subMapResults = map.subMap(prefix, true, prefixEnd, false);
+        }
+
+        NavigableMap<Bytes, byte[]> subMapExpected = new TreeMap<>();
+        subMapExpected.put(key1, val);
+        subMapExpected.put(key2, val);
+        subMapExpected.put(key3, val);
+
+        assertEquals(subMapExpected.keySet(), subMapResults.keySet());
+    }
+}

core/src/main/scala/kafka/tools/StreamsResetter.java

@@ -675,7 +675,9 @@ private boolean isInternalTopic(final String topicName) {
         // Cf. https://issues.apache.org/jira/browse/KAFKA-7930
         return !isInputTopic(topicName) && !isIntermediateTopic(topicName)
             && topicName.startsWith(options.valueOf(applicationIdOption) + "-")
-            && (topicName.endsWith("-changelog") || topicName.endsWith("-repartition"));
+            && (topicName.endsWith("-changelog") || topicName.endsWith("-repartition")
+                || topicName.endsWith("-subscription-registration-topic")
+                || topicName.endsWith("-subscription-response-topic"));
     }
 
     public static void main(final String[] args) {

streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java

@@ -30,6 +30,8 @@
 import org.apache.kafka.streams.state.QueryableStoreType;
 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
 
+import java.util.function.Function;
+
 /**
  * {@code KTable} is an abstraction of a <i>changelog stream</i> from a primary-keyed table.
  * Each record in this changelog stream is an update on the primary-keyed table with the record key as the primary key.
@@ -2117,6 +2119,92 @@ <VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,
                                      final Named named,
                                      final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
+    /**
+     *
+     * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed inner join. Records from this
+     * table are joined according to the result of keyExtractor on the other KTable.
+     *
+     * @param other  the other {@code KTable} to be joined with this {@code KTable}. Keyed by KO.
+     * @param foreignKeyExtractor a {@link Function} that extracts the key (KO) from this table's value (V)
+     * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
+     * @param named     a {@link Named} config used to name the processor in the topology
+     * @param materialized  a {@link Materialized} that describes how the {@link StateStore} for the resulting {@code KTable}
+     *                      should be materialized. Cannot be {@code null}
+     * @param <VR> the value type of the result {@code KTable}
+     * @param <KO> the key type of the other {@code KTable}
+     * @param <VO> the value type of the other {@code KTable}
+     * @return
+     */
+    <VR, KO, VO> KTable<K, VR> join(final KTable<KO, VO> other,
+                                    final Function<V, KO> foreignKeyExtractor,
+                                    final ValueJoiner<V, VO, VR> joiner,
+                                    final Named named,
+                                    final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+
+    /**
+     *
+     * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed inner join. Records from this
+     * table are joined according to the result of keyExtractor on the other KTable.
+     *
+     * @param other  the other {@code KTable} to be joined with this {@code KTable}. Keyed by KO.
+     * @param foreignKeyExtractor a {@link Function} that extracts the key (KO) from this table's value (V)
+     * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
+     * @param materialized  a {@link Materialized} that describes how the {@link StateStore} for the resulting {@code KTable}
+     *                      should be materialized. Cannot be {@code null}
+     * @param <VR> the value type of the result {@code KTable}
+     * @param <KO> the key type of the other {@code KTable}
+     * @param <VO> the value type of the other {@code KTable}
+     * @return
+     */
+    <VR, KO, VO> KTable<K, VR> join(final KTable<KO, VO> other,
+                                    final Function<V, KO> foreignKeyExtractor,
+                                    final ValueJoiner<V, VO, VR> joiner,
+                                    final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+
+    /**
+     *
+     * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed left join. Records from this
+     * table are joined according to the result of keyExtractor on the other KTable.
+     *
+     * @param other  the other {@code KTable} to be joined with this {@code KTable}. Keyed by KO.
+     * @param foreignKeyExtractor a {@link Function} that extracts the key (KO) from this table's value (V). If the
+     *      *                            resultant foreignKey is null, the record will not propagate to the output.
+     * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
+     * @param named     a {@link Named} config used to name the processor in the topology
+     * @param materialized  a {@link Materialized} that describes how the {@link StateStore} for the resulting {@code KTable}
+     *                      should be materialized. Cannot be {@code null}
+     * @param <VR> the value type of the result {@code KTable}
+     * @param <KO> the key type of the other {@code KTable}
+     * @param <VO> the value type of the other {@code KTable}
+     * @return a {@code KTable} that contains only those records that satisfy the given predicate
+     */
+    <VR, KO, VO> KTable<K, VR> leftJoin(final KTable<KO, VO> other,
+                                        final Function<V, KO> foreignKeyExtractor,
+                                        final ValueJoiner<V, VO, VR> joiner,
+                                        final Named named,
+                                        final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+
+    /**
+     *
+     * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed left join. Records from this
+     * table are joined according to the result of keyExtractor on the other KTable.
+     *
+     * @param other  the other {@code KTable} to be joined with this {@code KTable}. Keyed by KO.
+     * @param foreignKeyExtractor a {@link Function} that extracts the key (KO) from this table's value (V). If the
+     *                            resultant foreignKey is null, the record will not propagate to the output.
+     * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
+     * @param materialized  a {@link Materialized} that describes how the {@link StateStore} for the resulting {@code KTable}
+     *                      should be materialized. Cannot be {@code null}
+     * @param <VR> the value type of the result {@code KTable}
+     * @param <KO> the key type of the other {@code KTable}
+     * @param <VO> the value type of the other {@code KTable}
+     * @return a {@code KTable} that contains only those records that satisfy the given predicate
+     */
+    <VR, KO, VO> KTable<K, VR> leftJoin(final KTable<KO, VO> other,
+                                        final Function<V, KO> foreignKeyExtractor,
+                                        final ValueJoiner<V, VO, VR> joiner,
+                                        final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+
     /**
      * Get the name of the local state store used that can be used to query this {@code KTable}.
      *