public synchronized <K, V> KTable<K, V> table(final String topic,
                                                  final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized) {
        Objects.requireNonNull(topic, "topic can't be null");
        Objects.requireNonNull(materialized, "materialized can't be null");

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

        final ConsumedInternal<K, V> consumedInternal =
                new ConsumedInternal<>(Consumed.with(materializedInternal.keySerde(), materializedInternal.valueSerde()));

        return internalStreamsBuilder.table(topic, consumedInternal, materializedInternal);
    }
Create a {@link KTable} for the specified topic.
The default {@code "auto.offset.reset"} strategy as specified in the {@link StreamsConfig config} are used.
Key and value deserializers as defined by the options in {@link Materialized} 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} using the {@link 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}.

@param topic         the topic name; 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