fix incorrect public method name

Author: mjsax

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java

@@ -1622,7 +1622,7 @@ private void shouldMigrateSessionStoreToSessionStoreWithHeaders(final boolean is
         final StreamsBuilder newBuilder = new StreamsBuilder();
         final AtomicReference<SessionWithHeadersProcessor> processorRef = new AtomicReference<>();
         newBuilder.addStateStore(
-                Stores.sessionStoreBuilderWithHeaders(
+                Stores.sessionStoreWithHeadersBuilder(
                     isPersistent ? Stores.persistentSessionStoreWithHeaders(SESSION_STORE_NAME, Duration.ofMillis(RETENTION_MS)) :
                         Stores.inMemorySessionStore(SESSION_STORE_NAME, Duration.ofMillis(RETENTION_MS)),
                     Serdes.String(),
@@ -1680,7 +1680,7 @@ public void shouldProxySessionStoreToSessionStoreWithHeaders() throws Exception
         final StreamsBuilder newBuilder = new StreamsBuilder();
         final AtomicReference<SessionWithHeadersProcessor> processorRef = new AtomicReference<>();
         newBuilder.addStateStore(
-                Stores.sessionStoreBuilderWithHeaders(
+                Stores.sessionStoreWithHeadersBuilder(
                     Stores.persistentSessionStore(SESSION_STORE_NAME, Duration.ofMillis(RETENTION_MS)),  // non-headers supplier!
                     Serdes.String(),
                     Serdes.String()))
@@ -1941,7 +1941,7 @@ private void setupAndPopulateSessionStoreWithHeaders(final Properties props) thr
         final StreamsBuilder headersBuilder = new StreamsBuilder();
         final AtomicReference<SessionWithHeadersProcessor> processorRef = new AtomicReference<>();
         headersBuilder.addStateStore(
-                Stores.sessionStoreBuilderWithHeaders(
+                Stores.sessionStoreWithHeadersBuilder(
                     Stores.persistentSessionStoreWithHeaders(SESSION_STORE_NAME, Duration.ofMillis(RETENTION_MS)),
                     Serdes.String(),
                     Serdes.String()))

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

@@ -70,7 +70,7 @@ public  StoreBuilder<?> builder() {
 
         final StoreBuilder<?> builder;
         if (supplier instanceof HeadersBytesStoreSupplier) {
-            builder = Stores.sessionStoreBuilderWithHeaders(
+            builder = Stores.sessionStoreWithHeadersBuilder(
                 supplier,
                 materialized.keySerde(),
                 materialized.valueSerde()

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreBuilderWrapper.java

@@ -19,7 +19,7 @@
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.internals.SessionStoreBuilder;
-import org.apache.kafka.streams.state.internals.SessionStoreBuilderWithHeaders;
+import org.apache.kafka.streams.state.internals.SessionStoreWithHeadersBuilder;
 import org.apache.kafka.streams.state.internals.TimestampedWindowStoreBuilder;
 import org.apache.kafka.streams.state.internals.TimestampedWindowStoreWithHeadersBuilder;
 import org.apache.kafka.streams.state.internals.VersionedKeyValueStoreBuilder;
@@ -75,8 +75,8 @@ public long retentionPeriod() {
             return ((TimestampedWindowStoreWithHeadersBuilder<?, ?>) builder).retentionPeriod();
         } else if (builder instanceof SessionStoreBuilder) {
             return ((SessionStoreBuilder<?, ?>) builder).retentionPeriod();
-        } else if (builder instanceof SessionStoreBuilderWithHeaders) {
-            return ((SessionStoreBuilderWithHeaders<?, ?>) builder).retentionPeriod();
+        } else if (builder instanceof SessionStoreWithHeadersBuilder) {
+            return ((SessionStoreWithHeadersBuilder<?, ?>) builder).retentionPeriod();
         } else {
             throw new IllegalStateException("retentionPeriod is not supported when not a window store");
         }
@@ -112,7 +112,7 @@ public boolean isWindowStore() {
             || builder instanceof TimestampedWindowStoreBuilder
             || builder instanceof TimestampedWindowStoreWithHeadersBuilder
             || builder instanceof SessionStoreBuilder
-            || builder instanceof SessionStoreBuilderWithHeaders;
+            || builder instanceof SessionStoreWithHeadersBuilder;
     }
 
     @Override

streams/src/main/java/org/apache/kafka/streams/state/Stores.java

@@ -31,7 +31,7 @@
 import org.apache.kafka.streams.state.internals.RocksDbVersionedKeyValueBytesStoreSupplier;
 import org.apache.kafka.streams.state.internals.RocksDbWindowBytesStoreSupplier;
 import org.apache.kafka.streams.state.internals.SessionStoreBuilder;
-import org.apache.kafka.streams.state.internals.SessionStoreBuilderWithHeaders;
+import org.apache.kafka.streams.state.internals.SessionStoreWithHeadersBuilder;
 import org.apache.kafka.streams.state.internals.TimestampedKeyValueStoreBuilder;
 import org.apache.kafka.streams.state.internals.TimestampedKeyValueStoreBuilderWithHeaders;
 import org.apache.kafka.streams.state.internals.TimestampedWindowStoreBuilder;
@@ -658,12 +658,12 @@ public static <K, V> StoreBuilder<SessionStore<K, V>> sessionStoreBuilder(final
      * @param <V>           value type
      * @return an instance of {@link StoreBuilder} than can build a {@link SessionStoreWithHeaders}
      */
-    public static <K, V> StoreBuilder<SessionStoreWithHeaders<K, V>> sessionStoreBuilderWithHeaders(
+    public static <K, V> StoreBuilder<SessionStoreWithHeaders<K, V>> sessionStoreWithHeadersBuilder(
             final SessionBytesStoreSupplier supplier,
             final Serde<K> keySerde,
             final Serde<V> valueSerde) {
         Objects.requireNonNull(supplier, "supplier cannot be null");
-        return new SessionStoreBuilderWithHeaders<>(supplier, keySerde, valueSerde, Time.SYSTEM);
+        return new SessionStoreWithHeadersBuilder<>(supplier, keySerde, valueSerde, Time.SYSTEM);
     }
 
     /**

…nals/SessionStoreBuilderWithHeaders.java …nals/SessionStoreWithHeadersBuilder.javastreams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreBuilderWithHeaders.java renamed to streams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreWithHeadersBuilder.java streams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreBuilderWithHeaders.java renamed to streams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreWithHeadersBuilder.java

@@ -37,12 +37,12 @@
  * {@link AggregationWithHeaders} as the value wrapper and wires up the
  * header-aware store stack (change-logging, caching, metering).
  */
-public class SessionStoreBuilderWithHeaders<K, V>
+public class SessionStoreWithHeadersBuilder<K, V>
     extends AbstractStoreBuilder<K, AggregationWithHeaders<V>, SessionStoreWithHeaders<K, V>> {
 
     private final SessionBytesStoreSupplier storeSupplier;
 
-    public SessionStoreBuilderWithHeaders(final SessionBytesStoreSupplier storeSupplier,
+    public SessionStoreWithHeadersBuilder(final SessionBytesStoreSupplier storeSupplier,
                                           final Serde<K> keySerde,
                                           final Serde<V> valueSerde,
                                           final Time time) {

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

@@ -154,7 +154,7 @@ private void initStore(final boolean enableCaching) {
             Stores.persistentSessionStore(STORE_NAME, ofMillis(GAP_MS * 3));
 
         final StoreBuilder<SessionStoreWithHeaders<String, Long>> storeBuilder =
-            Stores.sessionStoreBuilderWithHeaders(supplier, Serdes.String(), Serdes.Long())
+            Stores.sessionStoreWithHeadersBuilder(supplier, Serdes.String(), Serdes.Long())
                 .withLoggingDisabled();
 
         if (enableCaching && emitStrategy.type() != EmitStrategy.StrategyType.ON_WINDOW_CLOSE) {

streams/src/test/java/org/apache/kafka/streams/state/QueryableStoreTypesWithHeadersTest.java

@@ -200,7 +200,7 @@ public void shouldAcceptRegularWindowStoreForWindowStoreType() {
     @Test
     public void shouldAcceptSessionStoreWithHeadersForSessionStoreType() {
         final SessionStoreWithHeaders<String, String> store =
-            Stores.sessionStoreBuilderWithHeaders(
+            Stores.sessionStoreWithHeadersBuilder(
                 Stores.inMemorySessionStore(
                     "test-session-store",
                     Duration.ofMillis(100)),

streams/src/test/java/org/apache/kafka/streams/state/internals/SessionStoreWithHeadersSerializerSideEffectTest.java

@@ -135,7 +135,7 @@ public void shouldPropagateSerializerHeaderSideEffectToChangelog() {
 
         // Create a session store with headers using our custom serializer
         builder.addStateStore(
-            Stores.sessionStoreBuilderWithHeaders(
+            Stores.sessionStoreWithHeadersBuilder(
                 Stores.inMemorySessionStore(
                     STORE_NAME,
                     Duration.ofMillis(10000L)