public synchronized <K, V> KTable<K, V> table(final String topic,
                                                  final Consumed<K, V> consumed) {
        Objects.requireNonNull(topic, "topic can't be null");
        Objects.requireNonNull(consumed, "consumed can't be null");
        final ConsumedInternal<K, V> consumedInternal = new ConsumedInternal<>(consumed);

        final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materializedInternal =
            new MaterializedInternal<>(
                Materialized.with(consumedInternal.keySerde(), consumedInternal.valueSerde()),
                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 topics 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} with an internal
store name. Note that store name may not be queryable through Interactive Queries.
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}.

@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}
@return a {@link KTable} for the specified topic
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java