public synchronized <K, V> KTable<K, V> table(final String topic,
                                                  final Consumed<K, V> consumed,
                                                  final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized) {
        Objects.requireNonNull(topic, "topic can't be null");
        Objects.requireNonNull(consumed, "consumed can't be null");
        Objects.requireNonNull(materialized, "materialized can't be null");
        final ConsumedInternal<K, V> consumedInternal = new ConsumedInternal<>(consumed);
        materialized.withKeySerde(consumedInternal.keySerde()).withValueSerde(consumedInternal.valueSerde());

        final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materializedInternal =
            new MaterializedInternal<>(materialized, internalStreamsBuilder, topic + "-");

        return internalStreamsBuilder.table(topic, consumedInternal, materializedInternal);
    }
Create a {@link KTable} for the specified topic.
The {@code "auto.offset.reset"} strategy, {@link TimestampExtractor}, key and value deserializers
are defined by the options in {@link Consumed} are used.
Input {@link KeyValue records} with {@code null} key will be dropped.
<p>
Note that the specified input topic must be partitioned by key.
If this is not the case the returned {@link KTable} will be corrupted.
<p>
The resulting {@link KTable} will be materialized in a local {@link KeyValueStore} using the given
{@code Materialized} instance.
An internal changelog topic is created by default. Because the source topic can
be used for recovery, you can avoid creating the changelog topic by setting
the {@code "topology.optimization"} to {@code "all"} in the {@link StreamsConfig}.
<p>
You should only specify serdes in the {@link Consumed} instance as these will also be used to overwrite the
serdes in {@link Materialized}, i.e.,
<pre> {@code
streamBuilder.table(topic, Consumed.with(Serde.String(), Serde.String()), Materialized.<String, String, KeyValueStore<Bytes, byte[]>as(storeName))
}
</pre>
To query the local {@link ReadOnlyKeyValueStore} it must be obtained via
{@link KafkaStreams#store(StoreQueryParameters) KafkaStreams#store(...)}:
<pre>{@code
KafkaStreams streams = ...
StoreQueryParameters<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>> storeQueryParams = StoreQueryParameters.fromNameAndType(queryableStoreName, QueryableStoreTypes.timestampedKeyValueStore());
ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>> localStore = streams.store(storeQueryParams);
K key = "some-key";
ValueAndTimestamp<V> valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
}</pre>
For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#metadataForAllStreamsClients()} to
query the value of the key on a parallel running instance of your Kafka Streams application.

@param topic              the topic name; cannot be {@code null}
@param consumed           the instance of {@link Consumed} used to define optional parameters; cannot be {@code null}
@param materialized       the instance of {@link Materialized} used to materialize a state store; cannot be {@code null}
@return a {@link KTable} for the specified topic
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java