payload: {"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/322919371","pull_request_review_id":284290098,"id":322919371,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDMyMjkxOTM3MQ==","diff_hunk":"@@ -0,0 +1,699 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements. See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License. You may obtain a copy of the License at\n+ *\n+ *    http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+package org.apache.kafka.streams.integration;\n+\n+import kafka.utils.MockTime;\n+import org.apache.kafka.clients.consumer.ConsumerConfig;\n+import org.apache.kafka.clients.producer.ProducerConfig;\n+import org.apache.kafka.common.serialization.FloatSerializer;\n+import org.apache.kafka.common.serialization.IntegerDeserializer;\n+import org.apache.kafka.common.serialization.IntegerSerializer;\n+import org.apache.kafka.common.serialization.LongSerializer;\n+import org.apache.kafka.common.serialization.Serdes;\n+import org.apache.kafka.common.serialization.StringDeserializer;\n+import org.apache.kafka.common.serialization.StringSerializer;\n+import org.apache.kafka.common.utils.Bytes;\n+import org.apache.kafka.streams.KafkaStreams;\n+import org.apache.kafka.streams.KeyValue;\n+import org.apache.kafka.streams.StreamsBuilder;\n+import org.apache.kafka.streams.StreamsConfig;\n+import org.apache.kafka.streams.Topology;\n+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;\n+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;\n+import org.apache.kafka.streams.kstream.Consumed;\n+import org.apache.kafka.streams.kstream.KTable;\n+import org.apache.kafka.streams.kstream.Materialized;\n+import org.apache.kafka.streams.kstream.Named;\n+import org.apache.kafka.streams.kstream.Produced;\n+import org.apache.kafka.streams.kstream.ValueJoiner;\n+import org.apache.kafka.streams.state.KeyValueIterator;\n+import org.apache.kafka.streams.state.KeyValueStore;\n+import org.apache.kafka.streams.state.QueryableStoreTypes;\n+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;\n+import org.apache.kafka.test.IntegrationTest;\n+import org.apache.kafka.test.TestUtils;\n+import org.junit.After;\n+import org.junit.Before;\n+import org.junit.BeforeClass;\n+import org.junit.ClassRule;\n+import org.junit.Test;\n+import org.junit.experimental.categories.Category;\n+\n+import java.io.IOException;\n+import java.util.Arrays;\n+import java.util.HashMap;\n+import java.util.HashSet;\n+import java.util.LinkedList;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Properties;\n+import java.util.Set;\n+import java.util.function.Function;\n+\n+import static org.junit.Assert.assertEquals;\n+import static org.junit.Assert.assertTrue;\n+import static org.junit.Assert.assertArrayEquals;\n+\n+\n+@Category({IntegrationTest.class})\n+public class KTableKTableForeignKeyJoinIntegrationTest {\n+    private final static int NUM_BROKERS = 1;\n+\n+    @ClassRule\n+    public final static EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);\n+    private final static MockTime MOCK_TIME = CLUSTER.time;\n+    private final static String LEFT_TABLE = \"left_table\";\n+    private final static String RIGHT_TABLE = \"right_table\";\n+    private final static String OUTPUT = \"output-topic\";\n+    private static Properties streamsConfig;\n+    private KafkaStreams streams;\n+    private KafkaStreams streamsTwo;\n+    private KafkaStreams streamsThree;\n+    private static final  Properties CONSUMER_CONFIG = new Properties();\n+    private static final Properties LEFT_PROD_CONF = new Properties();\n+    private static final Properties RIGHT_PROD_CONF = new Properties();\n+\n+    @BeforeClass\n+    public static void beforeTest() {\n+        //Use multiple partitions to ensure distribution of keys.\n+        LEFT_PROD_CONF.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());\n+        LEFT_PROD_CONF.put(ProducerConfig.ACKS_CONFIG, \"all\");\n+        LEFT_PROD_CONF.put(ProducerConfig.RETRIES_CONFIG, 0);\n+        LEFT_PROD_CONF.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);\n+        LEFT_PROD_CONF.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, FloatSerializer.class);\n+\n+        RIGHT_PROD_CONF.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());\n+        RIGHT_PROD_CONF.put(ProducerConfig.ACKS_CONFIG, \"all\");\n+        RIGHT_PROD_CONF.put(ProducerConfig.RETRIES_CONFIG, 0);\n+        RIGHT_PROD_CONF.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);\n+        RIGHT_PROD_CONF.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, LongSerializer.class);\n+\n+        streamsConfig = new Properties();\n+        streamsConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());\n+        streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, \"earliest\");\n+        streamsConfig.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());\n+        streamsConfig.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);\n+        streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);\n+\n+        CONSUMER_CONFIG.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());\n+        CONSUMER_CONFIG.put(ConsumerConfig.GROUP_ID_CONFIG, \"ktable-ktable-consumer\");\n+        CONSUMER_CONFIG.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);\n+        CONSUMER_CONFIG.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);\n+    }\n+\n+    @Before\n+    public void before() throws IOException, InterruptedException {\n+        CLUSTER.deleteTopicsAndWait(LEFT_TABLE);\n+        CLUSTER.deleteTopicsAndWait(RIGHT_TABLE);\n+        CLUSTER.deleteTopicsAndWait(OUTPUT);\n+\n+        CLUSTER.createTopic(LEFT_TABLE, 3, 1);\n+        CLUSTER.createTopic(RIGHT_TABLE, 3, 1);\n+        CLUSTER.createTopic(OUTPUT, 3, 1);\n+\n+        IntegrationTestUtils.purgeLocalStreamsState(streamsConfig);\n+    }\n+\n+    @After\n+    public void after() throws IOException {\n+        if (streams != null) {\n+            streams.close();\n+            streams = null;\n+        }\n+        if (streamsTwo != null) {\n+            streamsTwo.close();\n+            streamsTwo = null;\n+        }\n+        if (streamsThree != null) {\n+            streamsThree.close();\n+            streamsThree = null;\n+        }\n+        IntegrationTestUtils.purgeLocalStreamsState(streamsConfig);\n+    }\n+\n+    @Test\n+    public void doInnerJoinFromLeftThenDeleteLeftEntity() throws Exception {\n+        final List<KeyValue<String, Long>> rightTableEvents = Arrays.asList(new KeyValue<>(\"1\", 10L), new KeyValue<>(\"2\", 20L)); //partition 0\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+\n+        final String currentMethodName = new Object() { }\n+                .getClass()\n+                .getEnclosingMethod()\n+                .getName();\n+        createAndStartStreamsApplication(currentMethodName, false);\n+\n+        final List<KeyValue<Integer, Float>> leftTableEvents = Arrays.asList(new KeyValue<>(1, 1.33f), new KeyValue<>(2, 2.77f));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+\n+        final Set<KeyValue<Integer, String>> expected = new HashSet<>();\n+        expected.add(new KeyValue<>(1, \"value1=1.33,value2=10\"));\n+        expected.add(new KeyValue<>(2, \"value1=2.77,value2=20\"));\n+\n+        final Set<KeyValue<Integer, String>> result = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size()));\n+        assertEquals(expected, result);\n+\n+        //Now delete one LHS entity such that one delete is propagated down to the output.\n+        final Set<KeyValue<Integer, String>> expectedDeleted = new HashSet<>();\n+        expectedDeleted.add(new KeyValue<>(1, null));\n+\n+        final List<KeyValue<Integer, Float>> rightTableDeleteEvents = Arrays.asList(new KeyValue<>(1, null));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, rightTableDeleteEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        final Set<KeyValue<Integer, String>> resultDeleted = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expectedDeleted.size()));\n+        assertEquals(expectedDeleted, resultDeleted);\n+\n+        //Ensure the state stores have the correct values within:\n+        final Set<KeyValue<Integer, String>> expMatResults = new HashSet<>();\n+        expMatResults.add(new KeyValue<>(2, \"value1=2.77,value2=20\"));\n+        validateQueryableStoresContainExpectedKeyValues(expMatResults, currentMethodName);\n+    }\n+\n+    @Test\n+    public void doLeftJoinFromLeftThenDeleteLeftEntity() throws Exception {\n+        final List<KeyValue<String, Long>> rightTableEvents = Arrays.asList(new KeyValue<>(\"1\", 10L), new KeyValue<>(\"2\", 20L)); //partition 0\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+\n+        final String currentMethodName = new Object() { }\n+                .getClass()\n+                .getEnclosingMethod()\n+                .getName();\n+        createAndStartStreamsApplication(currentMethodName, true);\n+\n+        final List<KeyValue<Integer, Float>> leftTableEvents = Arrays.asList(new KeyValue<>(1, 1.33f), new KeyValue<>(2, 2.77f));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+\n+        final Set<KeyValue<Integer, String>> expected = new HashSet<>();\n+        expected.add(new KeyValue<>(1, \"value1=1.33,value2=10\"));\n+        expected.add(new KeyValue<>(2, \"value1=2.77,value2=20\"));\n+\n+        final Set<KeyValue<Integer, String>> result = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size()));\n+        assertEquals(expected, result);\n+\n+        //Now delete one LHS entity such that one delete is propagated down to the output.\n+        final Set<KeyValue<Integer, String>> expectedDeleted = new HashSet<>();\n+        expectedDeleted.add(new KeyValue<>(1, null));\n+\n+        final List<KeyValue<Integer, Float>> rightTableDeleteEvents = Arrays.asList(new KeyValue<>(1, null));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, rightTableDeleteEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        final Set<KeyValue<Integer, String>> resultDeleted = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expectedDeleted.size()));\n+        assertEquals(expectedDeleted, resultDeleted);\n+\n+        //Ensure the state stores have the correct values within:\n+        final Set<KeyValue<Integer, String>> expMatResults = new HashSet<>();\n+        expMatResults.add(new KeyValue<>(2, \"value1=2.77,value2=20\"));\n+        validateQueryableStoresContainExpectedKeyValues(expMatResults, currentMethodName);\n+    }\n+\n+    @Test\n+    public void doInnerJoinFromRightThenDeleteRightEntity() throws Exception {\n+        final List<KeyValue<Integer, Float>> leftTableEvents = Arrays.asList(\n+                new KeyValue<>(1, 1.33f),\n+                new KeyValue<>(2, 1.77f),\n+                new KeyValue<>(3, 3.77f));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        final String currentMethodName = new Object() { }\n+                .getClass()\n+                .getEnclosingMethod()\n+                .getName();\n+        createAndStartStreamsApplication(currentMethodName, false);\n+\n+        final List<KeyValue<String, Long>> rightTableEvents = Arrays.asList(\n+                new KeyValue<>(\"1\", 10L),  //partition 0\n+                new KeyValue<>(\"2\", 20L),  //partition 2\n+                new KeyValue<>(\"3\", 30L)); //partition 2\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+\n+        //Ensure that the joined values exist in the output\n+        final Set<KeyValue<Integer, String>> expected = new HashSet<>();\n+        expected.add(new KeyValue<>(1, \"value1=1.33,value2=10\"));   //Will be deleted.\n+        expected.add(new KeyValue<>(2, \"value1=1.77,value2=10\"));   //Will be deleted.\n+        expected.add(new KeyValue<>(3, \"value1=3.77,value2=30\"));   //Will not be deleted.\n+\n+        final Set<KeyValue<Integer, String>> result = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size()));\n+        assertEquals(expected, result);\n+\n+        //Now delete the RHS entity such that all matching keys have deletes propagated.\n+        final Set<KeyValue<Integer, String>> expectedDeleted = new HashSet<>();\n+        expectedDeleted.add(new KeyValue<>(1, null));\n+        expectedDeleted.add(new KeyValue<>(2, null));\n+\n+        final List<KeyValue<String, Long>> rightTableDeleteEvents = Arrays.asList(new KeyValue<>(\"1\", null));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableDeleteEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+        final Set<KeyValue<Integer, String>> resultDeleted = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expectedDeleted.size()));\n+        assertEquals(expectedDeleted, resultDeleted);\n+\n+        //Ensure the state stores have the correct values within:\n+        final Set<KeyValue<Integer, String>> expMatResults = new HashSet<>();\n+        expMatResults.add(new KeyValue<>(3, \"value1=3.77,value2=30\"));\n+        validateQueryableStoresContainExpectedKeyValues(expMatResults, currentMethodName);\n+    }\n+\n+    @Test\n+    public void doLeftJoinFromRightThenDeleteRightEntity() throws Exception {\n+        final List<KeyValue<Integer, Float>> leftTableEvents = Arrays.asList(\n+                new KeyValue<>(1, 1.33f),\n+                new KeyValue<>(2, 1.77f),\n+                new KeyValue<>(3, 3.77f));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        final String currentMethodName = new Object() { }\n+                .getClass()\n+                .getEnclosingMethod()\n+                .getName();\n+        createAndStartStreamsApplication(currentMethodName, true);\n+\n+        final List<KeyValue<String, Long>> rightTableEvents = Arrays.asList(\n+                new KeyValue<>(\"1\", 10L),  //partition 0\n+                new KeyValue<>(\"2\", 20L),  //partition 2\n+                new KeyValue<>(\"3\", 30L)); //partition 2\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+\n+        //Ensure that the joined values exist in the output\n+        final Set<KeyValue<Integer, String>> expected = new HashSet<>();\n+        expected.add(new KeyValue<>(1, \"value1=1.33,value2=10\"));   //Will be deleted.\n+        expected.add(new KeyValue<>(2, \"value1=1.77,value2=10\"));   //Will be deleted.\n+        expected.add(new KeyValue<>(3, \"value1=3.77,value2=30\"));   //Will not be deleted.\n+        //final HashSet<KeyValue<Integer, String>> expected = new HashSet<>(buildExpectedResults(leftTableEvents, rightTableEvents, false));\n+\n+        final Set<KeyValue<Integer, String>> result = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size()));\n+        assertEquals(expected, result);\n+\n+        //Now delete the RHS entity such that all matching keys have deletes propagated.\n+        //This will exercise the joiner with the RHS value == null.\n+        final Set<KeyValue<Integer, String>> expectedDeleted = new HashSet<>();\n+        expectedDeleted.add(new KeyValue<>(1, \"value1=1.33,value2=null\"));\n+        expectedDeleted.add(new KeyValue<>(2, \"value1=1.77,value2=null\"));\n+\n+        final List<KeyValue<String, Long>> rightTableDeleteEvents = Arrays.asList(new KeyValue<>(\"1\", null));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableDeleteEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+        final Set<KeyValue<Integer, String>> resultDeleted = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expectedDeleted.size()));\n+        assertEquals(expectedDeleted, resultDeleted);\n+\n+        //Ensure the state stores have the correct values within:\n+        final Set<KeyValue<Integer, String>> expMatResults = new HashSet<>();\n+        expMatResults.add(new KeyValue<>(1, \"value1=1.33,value2=null\"));\n+        expMatResults.add(new KeyValue<>(2, \"value1=1.77,value2=null\"));\n+        expMatResults.add(new KeyValue<>(3, \"value1=3.77,value2=30\"));\n+        validateQueryableStoresContainExpectedKeyValues(expMatResults, currentMethodName);\n+    }\n+\n+    @Test\n+    public void doInnerJoinProduceNullsWhenValueHasNonMatchingForeignKey() throws Exception {\n+        //There is no matching extracted foreign-key of 8 anywhere. Should not produce any output for INNER JOIN, only\n+        //because the state is transitioning from oldValue=null -> newValue=8.33.\n+        List<KeyValue<Integer, Float>> leftTableEvents = Arrays.asList(new KeyValue<>(1, 8.33f));\n+        final List<KeyValue<String, Long>> rightTableEvents = Arrays.asList(new KeyValue<>(\"1\", 10L)); //partition 0\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+\n+        final String currentMethodName = new Object() { }\n+                .getClass()\n+                .getEnclosingMethod()\n+                .getName();\n+        createAndStartStreamsApplication(currentMethodName, false);\n+\n+        //There is also no matching extracted foreign-key for 18 anywhere. This WILL produce a null output for INNER JOIN,\n+        //since we cannot remember (maintain state) that the FK=8 also produced a null result.\n+        leftTableEvents = Arrays.asList(new KeyValue<>(1, 18.00f));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+\n+        final List<KeyValue<Integer, String>> expected = new LinkedList<>();\n+        expected.add(new KeyValue<>(1, null));\n+\n+        final List<KeyValue<Integer, String>> result = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size());\n+        assertEquals(result, expected);\n+\n+        //Another change to FK that has no match on the RHS will result in another null\n+        leftTableEvents = Arrays.asList(new KeyValue<>(1, 100.00f));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        //Consume the next event - note that we are using the same consumerGroupId, so this will consume a new event.\n+        final List<KeyValue<Integer, String>> result2 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size());\n+        assertEquals(result2, expected);\n+\n+        //Now set the LHS event FK to match the rightTableEvents key-value.\n+        leftTableEvents = Arrays.asList(new KeyValue<>(1, 1.11f));\n+\n+        final List<KeyValue<Integer, String>> expected3 = new LinkedList<>();\n+        expected3.add(new KeyValue<>(1, \"value1=1.11,value2=10\"));\n+\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        final List<KeyValue<Integer, String>> result3 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected3.size());\n+        assertEquals(result3, expected3);\n+\n+        //Ensure the state stores have the correct values within:\n+        final Set<KeyValue<Integer, String>> expMatResults = new HashSet<>();\n+        expMatResults.add(new KeyValue<>(1, \"value1=1.11,value2=10\"));\n+        validateQueryableStoresContainExpectedKeyValues(expMatResults, currentMethodName);\n+    }\n+\n+    @Test\n+    public void doLeftJoinProduceJoinedResultsWhenValueHasNonMatchingForeignKey() throws Exception {\n+        //There is no matching extracted foreign-key of 8 anywhere.\n+        //However, it will still run the join function since this is LEFT join.\n+        List<KeyValue<Integer, Float>> leftTableEvents = Arrays.asList(new KeyValue<>(1, 8.33f));\n+        final List<KeyValue<String, Long>> rightTableEvents = Arrays.asList(new KeyValue<>(\"1\", 10L)); //partition 0\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+\n+        final String currentMethodName = new Object() { }\n+                .getClass()\n+                .getEnclosingMethod()\n+                .getName();\n+        createAndStartStreamsApplication(currentMethodName, true);\n+\n+        final List<KeyValue<Integer, String>> expected = new LinkedList<>();\n+        expected.add(new KeyValue<>(1, \"value1=8.33,value2=null\"));\n+        final List<KeyValue<Integer, String>> result = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size());\n+        assertEquals(expected, result);\n+\n+        //There is also no matching extracted foreign-key for 18 anywhere.\n+        //However, it will still run the join function since this if LEFT join.\n+        leftTableEvents = Arrays.asList(new KeyValue<>(1, 18.0f));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+\n+        final List<KeyValue<Integer, String>> expected2 = new LinkedList<>();\n+        expected2.add(new KeyValue<>(1, \"value1=18.0,value2=null\"));\n+        final List<KeyValue<Integer, String>> result2 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected2.size());\n+        assertEquals(expected2, result2);\n+\n+\n+        leftTableEvents = Arrays.asList(new KeyValue<>(1, 1.11f));\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+\n+        final List<KeyValue<Integer, String>> expected3 = new LinkedList<>();\n+        expected3.add(new KeyValue<>(1, \"value1=1.11,value2=10\"));\n+        final List<KeyValue<Integer, String>> result3 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected3.size());\n+        assertEquals(expected3, result3);\n+\n+        //Ensure the state stores have the correct values within:\n+        final Set<KeyValue<Integer, String>> expMatResults = new HashSet<>();\n+        expMatResults.add(new KeyValue<>(1, \"value1=1.11,value2=10\"));\n+        validateQueryableStoresContainExpectedKeyValues(expMatResults, currentMethodName);\n+    }\n+\n+    @Test\n+    public void doInnerJoinFilterOutRapidlyChangingForeignKeyValues() throws Exception {\n+        final List<KeyValue<Integer, Float>> leftTableEvents = Arrays.asList(\n+                new KeyValue<>(1, 1.33f),\n+                new KeyValue<>(2, 2.22f),\n+                new KeyValue<>(3, -1.22f), //Won't be joined in\n+                new KeyValue<>(4, -2.22f), //Won't be joined in\n+                new KeyValue<>(5, 2.22f)\n+        );\n+\n+        //Partitions pre-computed using the default Murmur2 hash, just to ensure that all 3 partitions will be exercised.\n+        final List<KeyValue<String, Long>> rightTableEvents = Arrays.asList(\n+                new KeyValue<>(\"0\", 0L),  //partition 2\n+                new KeyValue<>(\"1\", 10L), //partition 0\n+                new KeyValue<>(\"2\", 20L), //partition 2\n+                new KeyValue<>(\"3\", 30L), //partition 2\n+                new KeyValue<>(\"4\", 40L), //partition 1\n+                new KeyValue<>(\"5\", 50L), //partition 0\n+                new KeyValue<>(\"6\", 60L), //partition 1\n+                new KeyValue<>(\"7\", 70L), //partition 0\n+                new KeyValue<>(\"8\", 80L), //partition 0\n+                new KeyValue<>(\"9\", 90L)  //partition 2\n+        );\n+\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+\n+        final Set<KeyValue<Integer, String>> expected = new HashSet<>();\n+        expected.add(new KeyValue<>(1, \"value1=1.33,value2=10\"));\n+        expected.add(new KeyValue<>(2, \"value1=2.22,value2=20\"));\n+        expected.add(new KeyValue<>(5, \"value1=2.22,value2=20\"));\n+\n+        final String currentMethodName = new Object() { }\n+                .getClass()\n+                .getEnclosingMethod()\n+                .getName();\n+        createAndStartStreamsApplication(currentMethodName, false);\n+\n+        final Set<KeyValue<Integer, String>> result = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size()));\n+\n+        assertEquals(result, expected);\n+\n+        //Rapidly change the foreign key, to validate that the hashing prevents incorrect results from being output,\n+        //and that eventually the correct value is output.\n+        final List<KeyValue<Integer, Float>> table1ForeignKeyChange = Arrays.asList(\n+                new KeyValue<>(3, 2.22f), //Partition 2\n+                new KeyValue<>(3, 3.33f), //Partition 2\n+                new KeyValue<>(3, 4.44f), //Partition 1\n+                new KeyValue<>(3, 5.55f), //Partition 0\n+                new KeyValue<>(3, 9.99f), //Partition 2\n+                new KeyValue<>(3, 8.88f), //Partition 0\n+                new KeyValue<>(3, 0.23f), //Partition 2\n+                new KeyValue<>(3, 7.77f), //Partition 0\n+                new KeyValue<>(3, 6.66f), //Partition 1\n+                new KeyValue<>(3, 1.11f)  //Partition 0 - This will be the final result.\n+        );\n+\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, table1ForeignKeyChange, LEFT_PROD_CONF, MOCK_TIME);\n+        final List<KeyValue<Integer, String>> resultTwo = IntegrationTestUtils.readKeyValues(OUTPUT, CONSUMER_CONFIG, 15 * 1000L, Integer.MAX_VALUE);\n+\n+        final List<KeyValue<Integer, String>> expectedTwo = new LinkedList<>();\n+        expectedTwo.add(new KeyValue<>(3, \"value1=1.11,value2=10\"));\n+        assertArrayEquals(resultTwo.toArray(), expectedTwo.toArray());\n+\n+        //Ensure the state stores have the correct values within:\n+        final Set<KeyValue<Integer, String>> expMatResults = new HashSet<>();\n+        expMatResults.addAll(expected);\n+        expMatResults.addAll(expectedTwo);\n+        validateQueryableStoresContainExpectedKeyValues(expMatResults, currentMethodName);\n+    }\n+\n+    @Test\n+    public void doLeftJoinFilterOutRapidlyChangingForeignKeyValues() throws Exception {\n+        final List<KeyValue<Integer, Float>> leftTableEvents = Arrays.asList(\n+                new KeyValue<>(1, 1.33f),\n+                new KeyValue<>(2, 2.22f)\n+        );\n+\n+        //Partitions pre-computed using the default Murmur2 hash, just to ensure that all 3 partitions will be exercised.\n+        final List<KeyValue<String, Long>> rightTableEvents = Arrays.asList(\n+                new KeyValue<>(\"0\", 0L),  //partition 2\n+                new KeyValue<>(\"1\", 10L), //partition 0\n+                new KeyValue<>(\"2\", 20L), //partition 2\n+                new KeyValue<>(\"3\", 30L), //partition 2\n+                new KeyValue<>(\"4\", 40L), //partition 1\n+                new KeyValue<>(\"5\", 50L), //partition 0\n+                new KeyValue<>(\"6\", 60L), //partition 1\n+                new KeyValue<>(\"7\", 70L), //partition 0\n+                new KeyValue<>(\"8\", 80L), //partition 0\n+                new KeyValue<>(\"9\", 90L)  //partition 2\n+        );\n+\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, leftTableEvents, LEFT_PROD_CONF, MOCK_TIME);\n+        IntegrationTestUtils.produceKeyValuesSynchronously(RIGHT_TABLE, rightTableEvents, RIGHT_PROD_CONF, MOCK_TIME);\n+\n+        final Set<KeyValue<Integer, String>> expected = new HashSet<>();\n+        expected.add(new KeyValue<>(1, \"value1=1.33,value2=10\"));\n+        expected.add(new KeyValue<>(2, \"value1=2.22,value2=20\"));\n+\n+        final String currentMethodName = new Object() { }\n+                .getClass()\n+                .getEnclosingMethod()\n+                .getName();\n+        createAndStartStreamsApplication(currentMethodName, false);\n+\n+        final Set<KeyValue<Integer, String>> result = new HashSet<>(IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(\n+                CONSUMER_CONFIG,\n+                OUTPUT,\n+                expected.size()));\n+\n+        assertEquals(result, expected);\n+\n+        //Rapidly change the foreign key, to validate that the hashing prevents incorrect results from being output,\n+        //and that eventually the correct value is output.\n+        final List<KeyValue<Integer, Float>> table1ForeignKeyChange = Arrays.asList(\n+                new KeyValue<>(3, 2.22f), //Partition 2\n+                new KeyValue<>(3, 3.33f), //Partition 2\n+                new KeyValue<>(3, 4.44f), //Partition 1\n+                new KeyValue<>(3, 5.55f), //Partition 0\n+                new KeyValue<>(3, 9.99f), //Partition 2\n+                new KeyValue<>(3, 8.88f), //Partition 0\n+                new KeyValue<>(3, 0.23f), //Partition 2\n+                new KeyValue<>(3, 7.77f), //Partition 0\n+                new KeyValue<>(3, 6.66f), //Partition 1\n+                new KeyValue<>(3, 1.11f)  //Partition 0 - This will be the final result.\n+        );\n+\n+        IntegrationTestUtils.produceKeyValuesSynchronously(LEFT_TABLE, table1ForeignKeyChange, LEFT_PROD_CONF, MOCK_TIME);\n+        final List<KeyValue<Integer, String>> resultTwo = IntegrationTestUtils.readKeyValues(OUTPUT, CONSUMER_CONFIG, 15 * 1000L, Integer.MAX_VALUE);\n+\n+        final List<KeyValue<Integer, String>> expectedTwo = new LinkedList<>();\n+        expectedTwo.add(new KeyValue<>(3, \"value1=1.11,value2=10\"));\n+\n+        assertArrayEquals(resultTwo.toArray(), expectedTwo.toArray());\n+\n+        //Ensure the state stores have the correct values within:\n+        final Set<KeyValue<Integer, String>> expMatResults = new HashSet<>();\n+        expMatResults.addAll(expected);\n+        expMatResults.addAll(expectedTwo);\n+        validateQueryableStoresContainExpectedKeyValues(expMatResults, currentMethodName);\n+    }\n+\n+    private void createAndStartStreamsApplication(final String queryableStoreName, final boolean leftJoin) {\n+        streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, \"ktable-ktable-joinOnForeignKey-\" + queryableStoreName);\n+        streams = prepareTopology(queryableStoreName, leftJoin);\n+        streamsTwo = prepareTopology(queryableStoreName, leftJoin);\n+        streamsThree = prepareTopology(queryableStoreName, leftJoin);\n+        streams.start();\n+        streamsTwo.start();\n+        streamsThree.start();\n+    }\n+\n+    // These are hardwired into the test logic for readability sake.\n+    // Do not change unless you want to change all the test results as well.\n+    private ValueJoiner<Float, Long, String> joiner = (value1, value2) -> \"value1=\" + value1 + \",value2=\" + value2;\n+    //Do not change. See above comment.\n+    private Function<Float, String> tableOneKeyExtractor = value -> Integer.toString((int) value.floatValue());\n+\n+    private void validateQueryableStoresContainExpectedKeyValues(final Set<KeyValue<Integer, String>> expectedResult,\n+                                                                 final String queryableStoreName) {\n+        final ReadOnlyKeyValueStore<Integer, String> myJoinStoreOne = streams.store(queryableStoreName,\n+                QueryableStoreTypes.keyValueStore());\n+\n+        final ReadOnlyKeyValueStore<Integer, String> myJoinStoreTwo = streamsTwo.store(queryableStoreName,\n+                QueryableStoreTypes.keyValueStore());\n+\n+        final ReadOnlyKeyValueStore<Integer, String> myJoinStoreThree = streamsThree.store(queryableStoreName,\n+                QueryableStoreTypes.keyValueStore());\n+\n+        // store only keeps last set of values, not entire stream of value changes\n+        final Map<Integer, String> expectedInStore = new HashMap<>();\n+        for (final KeyValue<Integer, String> expected : expectedResult) {\n+            expectedInStore.put(expected.key, expected.value);\n+        }\n+\n+        // depending on partition assignment, the values will be in one of the three stream clients.\n+        for (final Map.Entry<Integer, String> expected : expectedInStore.entrySet()) {\n+            final String one = myJoinStoreOne.get(expected.getKey());\n+            final String two = myJoinStoreTwo.get(expected.getKey());\n+            final String three = myJoinStoreThree.get(expected.getKey());\n+\n+            String result;\n+            if (one != null)\n+                result = one;\n+            else if (two != null)\n+                result = two;\n+            else if (three != null)\n+                result = three;\n+            else\n+                throw new RuntimeException(\"Cannot find key \" + expected.getKey() + \" in any of the state stores\");\n+            assertEquals(expected.getValue(), result);\n+        }\n+\n+        //Merge all the iterators together to ensure that their sum equals the total set of expected elements.\n+        final KeyValueIterator<Integer, String> allOne = myJoinStoreOne.all();\n+        final KeyValueIterator<Integer, String> allTwo = myJoinStoreTwo.all();\n+        final KeyValueIterator<Integer, String> allThree = myJoinStoreThree.all();\n+\n+        final List<KeyValue<Integer, String>> all = new LinkedList<>();\n+\n+        while (allOne.hasNext()) {\n+            all.add(allOne.next());\n+        }\n+        while (allTwo.hasNext()) {\n+            all.add(allTwo.next());\n+        }\n+        while (allThree.hasNext()) {\n+            all.add(allThree.next());\n+        }\n+        allOne.close();\n+        allTwo.close();\n+        allThree.close();\n+\n+        for (final KeyValue<Integer, String> elem : all) {\n+            assertTrue(expectedResult.contains(elem));\n+        }\n+    }\n+\n+    private KafkaStreams prepareTopology(final String queryableStoreName, final boolean leftJoin) {\n+        final StreamsBuilder builder = new StreamsBuilder();\n+\n+        final KTable<Integer, Float> left = builder.table(LEFT_TABLE, Consumed.with(Serdes.Integer(), Serdes.Float()));\n+        final KTable<String, Long> right = builder.table(RIGHT_TABLE, Consumed.with(Serdes.String(), Serdes.Long()));\n+\n+        final Materialized<Integer, String, KeyValueStore<Bytes, byte[]>> materialized;\n+        if (queryableStoreName != null) {\n+            materialized = Materialized.<Integer, String, KeyValueStore<Bytes, byte[]>>as(queryableStoreName)\n+                    .withKeySerde(Serdes.Integer())\n+                    .withValueSerde(Serdes.String())\n+                    .withCachingDisabled();\n+        } else {\n+            throw new RuntimeException(\"Current implementation of join on foreign key requires a materialized store\");\n+        }\n+\n+        if (leftJoin)\n+            left.leftJoin(right, tableOneKeyExtractor, joiner, Named.as(\"customName\"), materialized)\n+                .toStream()\n+                .to(OUTPUT, Produced.with(Serdes.Integer(), Serdes.String()));\n+        else\n+            left.join(right, tableOneKeyExtractor, joiner/*, Named.as(\"customName\")*/, materialized)\n+                .toStream()\n+                .to(OUTPUT, Produced.with(Serdes.Integer(), Serdes.String()));\n+\n+        final Topology topology = builder.build();\n+        System.out.println(\"Topology description \" + topology.describe());","path":"streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyJoinIntegrationTest.java","position":695,"original_position":695,"commit_id":"9da1cd9a2450e1e0ca3964758d51291e68332ff8","original_commit_id":"9da1cd9a2450e1e0ca3964758d51291e68332ff8","user":{"login":"bbejeck","id":199238,"node_id":"MDQ6VXNlcjE5OTIzOA==","avatar_url":"https://avatars3.githubusercontent.com/u/199238?v=4","gravatar_id":"","url":"https://api.github.com/users/bbejeck","html_url":"https://github.com/bbejeck","followers_url":"https://api.github.com/users/bbejeck/followers","following_url":"https://api.github.com/users/bbejeck/following{/other_user}","gists_url":"https://api.github.com/users/bbejeck/gists{/gist_id}","starred_url":"https://api.github.com/users/bbejeck/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbejeck/subscriptions","organizations_url":"https://api.github.com/users/bbejeck/orgs","repos_url":"https://api.github.com/users/bbejeck/repos","events_url":"https://api.github.com/users/bbejeck/events{/privacy}","received_events_url":"https://api.github.com/users/bbejeck/received_events","type":"User","site_admin":false},"body":"leftover debugging?","created_at":"2019-09-10T19:22:00Z","updated_at":"2019-09-10T19:24:11Z","html_url":"https://github.com/apache/kafka/pull/5527#discussion_r322919371","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/5527","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/322919371"},"html":{"href":"https://github.com/apache/kafka/pull/5527#discussion_r322919371"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/5527"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/5527","id":209224098,"node_id":"MDExOlB1bGxSZXF1ZXN0MjA5MjI0MDk4","html_url":"https://github.com/apache/kafka/pull/5527","diff_url":"https://github.com/apache/kafka/pull/5527.diff","patch_url":"https://github.com/apache/kafka/pull/5527.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/5527","number":5527,"state":"open","locked":false,"title":"[REVIEW NEEDED] - KAFKA-3705 Added a foreignKeyJoin implementation for KTable.","user":{"login":"bellemare","id":8051579,"node_id":"MDQ6VXNlcjgwNTE1Nzk=","avatar_url":"https://avatars1.githubusercontent.com/u/8051579?v=4","gravatar_id":"","url":"https://api.github.com/users/bellemare","html_url":"https://github.com/bellemare","followers_url":"https://api.github.com/users/bellemare/followers","following_url":"https://api.github.com/users/bellemare/following{/other_user}","gists_url":"https://api.github.com/users/bellemare/gists{/gist_id}","starred_url":"https://api.github.com/users/bellemare/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bellemare/subscriptions","organizations_url":"https://api.github.com/users/bellemare/orgs","repos_url":"https://api.github.com/users/bellemare/repos","events_url":"https://api.github.com/users/bellemare/events{/privacy}","received_events_url":"https://api.github.com/users/bellemare/received_events","type":"User","site_admin":false},"body":"https://issues.apache.org/jira/browse/KAFKA-3705\r\n\r\nForeign Key Join:\r\n================================\r\nAllows for a KTable to map its value to a given foreign key and join on another KTable keyed on that foreign key. Applies the joiner, then returns the tuples keyed on the original key. This supports updates from both sides of the join.\r\n\r\nDesign Philosophy:\r\n================================\r\nThe intent of this design was to build a totally encapsulated function that operates very similarly to the regular join function. No further work is required by the user to obtain their foreignKeyJoin results after calling the function. That being said, there is increased cost in some of the topology components, especially due to resolving out-of-order arrival due to foreign key changes. I would appreciate any and all feedback on this approach, as my understanding of the Kafka Streams DSL is to provide higher level functionality without requiring the users to know exactly what's going on under the hood. \r\n\r\nSome points of note:\r\n1) Requires an additional materialized State Store for the prefixScanning of the repartitioned CombinedKey events.\r\n\r\n2) ReadOnlyKeyValueStore interface was modified to contain prefixScan. This requires that all implementations support this, but follows an existing precedent where some store functions are already stubbed out with exceptions.\r\n\r\n3) Currently limited to Inner Join (can do more join logic in future - just limiting the focus of this KIP).\r\n\r\n4) Application Reset does not seem to delete the new internal topics that I have added. (only tested with Kafka 1.0).\r\n\r\n5) Only works with identical number of input partitions at the moment, though it may be possible to get it working with KTables of varying input partition count.\r\n\r\nTesting:\r\n================================\r\nTesting is covered by a two integration tests that exercises the foreign key join. \r\nThe first test exercises the out-of-order resolution and partitioning strategies by running three streams instances on three partitions. This demonstrates the scalability of the proposed solution.\r\n\r\n*important* The second test (KTableKTableForeignKeyInnerJoinMultiIntegrationTest) attempts to join using foreign key twice. This results in a NullPointerException regarding a missing task, and must be resolved before committing this.\r\n","created_at":"2018-08-17T18:27:07Z","updated_at":"2019-09-10T19:24:11Z","closed_at":null,"merged_at":null,"merge_commit_sha":"d252b51417f49f726db5f0717764a358dbc32144","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[{"id":820747028,"node_id":"MDU6TGFiZWw4MjA3NDcwMjg=","url":"https://api.github.com/repos/apache/kafka/labels/streams","name":"streams","color":"63edea","default":false}],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/5527/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/5527/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/5527/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/9da1cd9a2450e1e0ca3964758d51291e68332ff8","head":{"label":"bellemare:trunk-oneToMany","ref":"trunk-oneToMany","sha":"9da1cd9a2450e1e0ca3964758d51291e68332ff8","user":{"login":"bellemare","id":8051579,"node_id":"MDQ6VXNlcjgwNTE1Nzk=","avatar_url":"https://avatars1.githubusercontent.com/u/8051579?v=4","gravatar_id":"","url":"https://api.github.com/users/bellemare","html_url":"https://github.com/bellemare","followers_url":"https://api.github.com/users/bellemare/followers","following_url":"https://api.github.com/users/bellemare/following{/other_user}","gists_url":"https://api.github.com/users/bellemare/gists{/gist_id}","starred_url":"https://api.github.com/users/bellemare/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bellemare/subscriptions","organizations_url":"https://api.github.com/users/bellemare/orgs","repos_url":"https://api.github.com/users/bellemare/repos","events_url":"https://api.github.com/users/bellemare/events{/privacy}","received_events_url":"https://api.github.com/users/bellemare/received_events","type":"User","site_admin":false},"repo":{"id":93407443,"node_id":"MDEwOlJlcG9zaXRvcnk5MzQwNzQ0Mw==","name":"kafka","full_name":"bellemare/kafka","private":false,"owner":{"login":"bellemare","id":8051579,"node_id":"MDQ6VXNlcjgwNTE1Nzk=","avatar_url":"https://avatars1.githubusercontent.com/u/8051579?v=4","gravatar_id":"","url":"https://api.github.com/users/bellemare","html_url":"https://github.com/bellemare","followers_url":"https://api.github.com/users/bellemare/followers","following_url":"https://api.github.com/users/bellemare/following{/other_user}","gists_url":"https://api.github.com/users/bellemare/gists{/gist_id}","starred_url":"https://api.github.com/users/bellemare/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bellemare/subscriptions","organizations_url":"https://api.github.com/users/bellemare/orgs","repos_url":"https://api.github.com/users/bellemare/repos","events_url":"https://api.github.com/users/bellemare/events{/privacy}","received_events_url":"https://api.github.com/users/bellemare/received_events","type":"User","site_admin":false},"html_url":"https://github.com/bellemare/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/bellemare/kafka","forks_url":"https://api.github.com/repos/bellemare/kafka/forks","keys_url":"https://api.github.com/repos/bellemare/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/bellemare/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/bellemare/kafka/teams","hooks_url":"https://api.github.com/repos/bellemare/kafka/hooks","issue_events_url":"https://api.github.com/repos/bellemare/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/bellemare/kafka/events","assignees_url":"https://api.github.com/repos/bellemare/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/bellemare/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/bellemare/kafka/tags","blobs_url":"https://api.github.com/repos/bellemare/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/bellemare/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/bellemare/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/bellemare/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/bellemare/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/bellemare/kafka/languages","stargazers_url":"https://api.github.com/repos/bellemare/kafka/stargazers","contributors_url":"https://api.github.com/repos/bellemare/kafka/contributors","subscribers_url":"https://api.github.com/repos/bellemare/kafka/subscribers","subscription_url":"https://api.github.com/repos/bellemare/kafka/subscription","commits_url":"https://api.github.com/repos/bellemare/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/bellemare/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/bellemare/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/bellemare/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/bellemare/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/bellemare/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/bellemare/kafka/merges","archive_url":"https://api.github.com/repos/bellemare/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/bellemare/kafka/downloads","issues_url":"https://api.github.com/repos/bellemare/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/bellemare/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/bellemare/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/bellemare/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/bellemare/kafka/labels{/name}","releases_url":"https://api.github.com/repos/bellemare/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/bellemare/kafka/deployments","created_at":"2017-06-05T13:36:25Z","updated_at":"2019-09-05T07:52:07Z","pushed_at":"2019-09-10T16:37:50Z","git_url":"git://github.com/bellemare/kafka.git","ssh_url":"git@github.com:bellemare/kafka.git","clone_url":"https://github.com/bellemare/kafka.git","svn_url":"https://github.com/bellemare/kafka","homepage":null,"size":84792,"stargazers_count":1,"watchers_count":1,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1,"mirror_url":null,"archived":false,"disabled":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":1,"open_issues":0,"watchers":1,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"d54285f0c0f1095bab0fa860e73f9258def397c2","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"node_id":"MDEwOlJlcG9zaXRvcnkyMjExMjQz","name":"kafka","full_name":"apache/kafka","private":false,"owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2019-09-10T17:03:53Z","pushed_at":"2019-09-10T19:13:58Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":87191,"stargazers_count":13528,"watchers_count":13528,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":7175,"mirror_url":null,"archived":false,"disabled":false,"open_issues_count":662,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":7175,"open_issues":662,"watchers":13528,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/5527"},"html":{"href":"https://github.com/apache/kafka/pull/5527"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/5527"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/5527/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/5527/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/5527/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/9da1cd9a2450e1e0ca3964758d51291e68332ff8"}},"author_association":"NONE"}}
other: {"actor":{"display_login":"bbejeck"}}