KAFKA-6265: Add #queryableStoreName() to GlobalKTable

A spinoff of original pull request #4340 for resolving conflicts.

Author: RichardYuSTUG <yohan.richard.yu2@gmail.com>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4413 from ConcurrencyPractitioner/kafka-6265-2

docs/streams/upgrade-guide.html

@@ -71,6 +71,13 @@
         to distinguish them from configurations of other clients that share the same config names.
     </p>
 
+    <p>
+	New method in <code>GlobalKTable</code>
+    </p>
+    <ul>
+	<li> A method has been provided such that it will return the store name associated with the <code>GlobalKTable</code> or <code>null</code> if the store name is non-queryable. </li>
+    </ul>
+
     <p>
         New methods in <code>KafkaStreams</code>:
     </p>

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

@@ -67,4 +67,10 @@ import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
  */
 @InterfaceStability.Evolving
 public interface GlobalKTable<K, V> {
+    /**
+     * Get the name of the local state store that can be used to query this {@code GlobalKTable}.
+     *
+     * @return the underlying state store name, or {@code null} if this {@code GlobalKTable} cannot be queried.
+     */
+    String queryableStoreName();
 }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/GlobalKTableImpl.java

@@ -21,13 +21,29 @@ import org.apache.kafka.streams.kstream.GlobalKTable;
 public class GlobalKTableImpl<K, V> implements GlobalKTable<K, V> {
 
     private final KTableValueGetterSupplier<K, V> valueGetterSupplier;
+    private final boolean queryable;
 
     public GlobalKTableImpl(final KTableValueGetterSupplier<K, V> valueGetterSupplier) {
         this.valueGetterSupplier = valueGetterSupplier;
+        this.queryable = true;
+    }
+    
+    public GlobalKTableImpl(final KTableValueGetterSupplier<K, V> valueGetterSupplier, 
+                            final boolean queryable) {
+        this.valueGetterSupplier = valueGetterSupplier;
+        this.queryable = queryable;
     }
 
     KTableValueGetterSupplier<K, V> valueGetterSupplier() {
         return valueGetterSupplier;
     }
 
+    @Override
+    public String queryableStoreName() {
+        if (!queryable) {
+            return null;
+        }
+        return valueGetterSupplier.storeNames()[0];
+    }
+
 }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java

@@ -158,7 +158,7 @@ public class InternalStreamsBuilder implements InternalNameProvider {
                                                topic,
                                                processorName,
                                                tableSource);
-        return new GlobalKTableImpl<>(new KTableSourceValueGetterSupplier<K, V>(storeBuilder.name()));
+        return new GlobalKTableImpl<>(new KTableSourceValueGetterSupplier<K, V>(storeBuilder.name()), materialized.isQueryable());
     }
 
     @Override

streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java

@@ -157,6 +157,32 @@ public class InternalStreamsBuilderTest {
         assertNull(table1.queryableStoreName());
     }
     
+    @Test
+    public void shouldBuildGlobalTableWithNonQueryableStoreName() throws Exception {
+        final GlobalKTable<String, String> table1 = builder.globalTable(
+            "topic2",
+            consumed,
+            new MaterializedInternal<>(
+                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(null, null),
+                builder,
+                storePrefix));
+
+        assertNull(table1.queryableStoreName());
+    }
+
+    @Test
+    public void shouldBuildGlobalTableWithQueryaIbleStoreName() throws Exception {
+        final GlobalKTable<String, String> table1 = builder.globalTable(
+            "topic2",
+            consumed,
+            new MaterializedInternal<>(
+                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("globalTable"),
+                builder,
+                storePrefix));
+
+        assertEquals("globalTable", table1.queryableStoreName());
+    }
+
     @Test
     public void shouldBuildSimpleGlobalTableTopology() throws Exception {
         builder.globalTable("table",