Made assignThread default with an empty implementation.

Removed empty overrides of the method.

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

@@ -382,10 +382,6 @@ public class IQv2IntegrationTest {
                             this.context = stateStoreContext;
                         }
 
-                        @Override
-                        public void assignThread() {
-                        }
-
                         @Override
                         public void flush() {
 

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

@@ -820,10 +820,6 @@ public class VersionedKeyValueStoreIntegrationTest {
                 );
             }
 
-            @Override
-            public void assignThread() {
-            }
-
             @Override
             public void flush() {
                 // do nothing

streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java

@@ -71,7 +71,16 @@ public interface StateStore {
      */
     void init(final StateStoreContext stateStoreContext, final StateStore root);
 
-    void assignThread();
+
+    /**
+     * Assigns the store to a stream thread.
+     * <p>
+     * This function is called from the final stream thread,
+     * thus can be used to initialize resources that might require to know the running thread, e.g. metrics.
+     * </p>
+     * To access the thread use {@link Thread#currentThread()}
+     */
+    default void assignThread() { }
 
     /**
      * Flush any cached data

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java

@@ -98,10 +98,6 @@ public class InMemoryKeyValueStore implements KeyValueStore<Bytes, byte[]> {
         this.context = stateStoreContext;
     }
 
-    @Override
-    public void assignThread() {
-    }
-
     @Override
     public boolean persistent() {
         return false;

streams/src/main/java/org/apache/kafka/streams/state/internals/MemoryLRUCache.java

@@ -111,10 +111,6 @@ public class MemoryLRUCache implements KeyValueStore<Bytes, byte[]> {
         this.context = stateStoreContext;
     }
 
-    @Override
-    public void assignThread() {
-    }
-
     @Override
     public boolean persistent() {
         return false;

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java

@@ -194,10 +194,6 @@ public class RocksDBTimeOrderedKeyValueBuffer<K, V> implements TimeOrderedKeyVal
         }
     }
 
-    @Override
-    public void assignThread() {
-    }
-
     @Override
     public void flush() {
         store.flush();

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

@@ -58,10 +58,6 @@ public class NoOpWindowStore implements ReadOnlyWindowStore, StateStore {
     @Override
     public void init(final StateStoreContext stateStoreContext, final StateStore root) {}
 
-    @Override
-    public void assignThread() {
-    }
-
     @Override
     public void flush() {
 

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

@@ -380,10 +380,6 @@ public class ReadOnlyWindowStoreStub<K, V> implements ReadOnlyWindowStore<K, V>,
     @Override
     public void init(final StateStoreContext stateStoreContext, final StateStore root) {}
 
-    @Override
-    public void assignThread() {
-    }
-
     @Override
     public void flush() {
     }

streams/src/test/java/org/apache/kafka/test/MockKeyValueStore.java

@@ -64,10 +64,6 @@ public class MockKeyValueStore implements KeyValueStore<Object, Object> {
         closed = false;
     }
 
-    @Override
-    public void assignThread() {
-    }
-
     @Override
     public void flush() {
         instanceLastFlushCount.set(GLOBAL_FLUSH_COUNTER.getAndIncrement());

streams/src/test/java/org/apache/kafka/test/NoOpReadOnlyStore.java

@@ -88,10 +88,6 @@ public class NoOpReadOnlyStore<K, V> implements ReadOnlyKeyValueStore<K, V>, Sta
         stateStoreContext.register(root, (k, v) -> { });
     }
 
-    @Override
-    public void assignThread() {
-    }
-
     @Override
     public void flush() {
         flushed = true;

streams/src/test/java/org/apache/kafka/test/ReadOnlySessionStoreStub.java

@@ -186,10 +186,6 @@ public class ReadOnlySessionStoreStub<K, V> implements ReadOnlySessionStore<K, V
     @Override
     public void init(StateStoreContext stateStoreContext, StateStore root) {}
 
-    @Override
-    public void assignThread() {
-    }
-
     @Override
     public void flush() {