diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceByKey.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceByKey.java
index b69de39bf85a3..bce501048aef9 100644
--- a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceByKey.java
+++ b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceByKey.java
@@ -24,13 +24,13 @@
 import cz.seznam.euphoria.core.client.flow.Flow;
 import cz.seznam.euphoria.core.client.functional.CombinableReduceFunction;
 import cz.seznam.euphoria.core.client.functional.ReduceFunction;
-import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.functional.UnaryFunction;
 import cz.seznam.euphoria.core.client.graph.DAG;
 import cz.seznam.euphoria.core.client.io.Context;
 import cz.seznam.euphoria.core.client.operator.state.ListStorage;
 import cz.seznam.euphoria.core.client.operator.state.ListStorageDescriptor;
 import cz.seznam.euphoria.core.client.operator.state.State;
+import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.operator.state.StorageProvider;
 import cz.seznam.euphoria.core.client.operator.state.ValueStorage;
 import cz.seznam.euphoria.core.client.operator.state.ValueStorageDescriptor;
@@ -287,7 +287,7 @@ static class CombiningReduceState<E>
           extends State<E, E>
           implements StateSupport.MergeFrom<CombiningReduceState<E>> {
 
-    static final class Factory<E> implements StateFactory<E, State<E, E>> {
+    static final class Factory<E> implements StateFactory<E, E, State<E, E>> {
       private final CombinableReduceFunction<E> r;
 
       Factory(CombinableReduceFunction<E> r) {
@@ -295,8 +295,8 @@ static final class Factory<E> implements StateFactory<E, State<E, E>> {
       }
 
       @Override
-      public State<E, E> apply(Context<E> ctx, StorageProvider storageProvider) {
-        return new CombiningReduceState<>(ctx, storageProvider, r);
+      public State<E, E> createState(Context<E> context, StorageProvider storageProvider) {
+        return new CombiningReduceState<>(context, storageProvider, r);
       }
     }
 
@@ -344,22 +344,22 @@ public void mergeFrom(CombiningReduceState<E> other) {
     }
   }
 
-  private static class NonCombiningReduceState<VALUE, OUT>
-          extends State<VALUE, OUT>
-          implements StateSupport.MergeFrom<NonCombiningReduceState<VALUE, OUT>> {
+  private static class NonCombiningReduceState<IN, OUT>
+          extends State<IN, OUT>
+          implements StateSupport.MergeFrom<NonCombiningReduceState<IN, OUT>> {
 
-    static final class Factory<VALUE, OUT>
-            implements StateFactory<OUT, NonCombiningReduceState<VALUE, OUT>> {
-      private final ReduceFunction<VALUE, OUT> r;
+    static final class Factory<IN, OUT>
+            implements StateFactory<IN, OUT, NonCombiningReduceState<IN, OUT>> {
+      private final ReduceFunction<IN, OUT> r;
 
-      Factory(ReduceFunction<VALUE, OUT> r) {
+      Factory(ReduceFunction<IN, OUT> r) {
         this.r = Objects.requireNonNull(r);
       }
 
       @Override
-      public NonCombiningReduceState<VALUE, OUT>
-      apply(Context<OUT> ctx, StorageProvider storageProvider) {
-        return new NonCombiningReduceState<>(ctx, storageProvider, r);
+      public NonCombiningReduceState<IN, OUT>
+      createState(Context<OUT> context, StorageProvider storageProvider) {
+        return new NonCombiningReduceState<>(context, storageProvider, r);
       }
     }
 
@@ -367,22 +367,22 @@ static final class Factory<VALUE, OUT>
     private static final ListStorageDescriptor STORAGE_DESC =
             ListStorageDescriptor.of("values", (Class) Object.class);
 
-    private final ReduceFunction<VALUE, OUT> reducer;
-    private final ListStorage<VALUE> reducibleValues;
+    private final ReduceFunction<IN, OUT> reducer;
+    private final ListStorage<IN> reducibleValues;
 
     NonCombiningReduceState(Context<OUT> context,
                             StorageProvider storageProvider,
-                            ReduceFunction<VALUE, OUT> reducer) {
+                            ReduceFunction<IN, OUT> reducer) {
       super(context);
       this.reducer = Objects.requireNonNull(reducer);
 
       @SuppressWarnings("unchecked")
-      ListStorage<VALUE> ls = storageProvider.getListStorage(STORAGE_DESC);
+      ListStorage<IN> ls = storageProvider.getListStorage(STORAGE_DESC);
       reducibleValues = ls;
     }
 
     @Override
-    public void add(VALUE element) {
+    public void add(IN element) {
       reducibleValues.add(element);
     }
 
@@ -398,7 +398,7 @@ public void close() {
     }
 
     @Override
-    public void mergeFrom(NonCombiningReduceState<VALUE, OUT> other) {
+    public void mergeFrom(NonCombiningReduceState<IN, OUT> other) {
       this.reducibleValues.addAll(other.reducibleValues.get());
     }
   }
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceStateByKey.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceStateByKey.java
index 35c9f4e98a72a..083eaa3e8b749 100644
--- a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceStateByKey.java
+++ b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceStateByKey.java
@@ -23,9 +23,9 @@
 import cz.seznam.euphoria.core.client.dataset.windowing.Windowing;
 import cz.seznam.euphoria.core.client.flow.Flow;
 import cz.seznam.euphoria.core.client.functional.CombinableReduceFunction;
-import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.functional.UnaryFunction;
 import cz.seznam.euphoria.core.client.operator.state.State;
+import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.util.Pair;
 
 import javax.annotation.Nullable;
@@ -109,7 +109,7 @@ public static class DatasetBuilder3<IN, KEY, VALUE> {
     }
     public <OUT, STATE extends State<VALUE, OUT>> DatasetBuilder4<
             IN, KEY, VALUE, OUT, STATE> stateFactory(
-            StateFactory<OUT, STATE> stateFactory) {
+            StateFactory<VALUE, OUT, STATE> stateFactory) {
       return new DatasetBuilder4<>(
           name, input, keyExtractor, valueExtractor, stateFactory);
     }
@@ -120,12 +120,12 @@ public static class DatasetBuilder4<
     private final Dataset<IN> input;
     private final UnaryFunction<IN, KEY> keyExtractor;
     private final UnaryFunction<IN, VALUE> valueExtractor;
-    private final StateFactory<OUT, STATE> stateFactory;
+    private final StateFactory<VALUE, OUT, STATE> stateFactory;
     DatasetBuilder4(String name,
                     Dataset<IN> input,
                     UnaryFunction<IN, KEY> keyExtractor,
                     UnaryFunction<IN, VALUE> valueExtractor,
-                    StateFactory<OUT, STATE> stateFactory)
+                    StateFactory<VALUE, OUT, STATE> stateFactory)
     {
       this.name = Objects.requireNonNull(name);
       this.input = Objects.requireNonNull(input);
@@ -148,14 +148,14 @@ public static class DatasetBuilder5<
     private final Dataset<IN> input;
     private final UnaryFunction<IN, KEY> keyExtractor;
     private final UnaryFunction<IN, VALUE> valueExtractor;
-    private final StateFactory<OUT, STATE> stateFactory;
+    private final StateFactory<VALUE, OUT, STATE> stateFactory;
     private final CombinableReduceFunction<STATE> stateCombiner;
 
     DatasetBuilder5(String name,
                     Dataset<IN> input,
                     UnaryFunction<IN, KEY> keyExtractor,
                     UnaryFunction<IN, VALUE> valueExtractor,
-                    StateFactory<OUT, STATE> stateFactory,
+                    StateFactory<VALUE, OUT, STATE> stateFactory,
                     CombinableReduceFunction<STATE> stateCombiner) {
       // initialize default partitioning according to input
       super(new DefaultPartitioning<>(input.getNumPartitions()));
@@ -199,7 +199,7 @@ public static class DatasetBuilder6<
     private final Dataset<IN> input;
     private final UnaryFunction<IN, KEY> keyExtractor;
     private final UnaryFunction<IN, VALUE> valueExtractor;
-    private final StateFactory<OUT, STATE> stateFactory;
+    private final StateFactory<VALUE, OUT, STATE> stateFactory;
     private final CombinableReduceFunction<STATE> stateCombiner;
     @Nullable
     private final Windowing<WIN, W> windowing;
@@ -210,7 +210,7 @@ public static class DatasetBuilder6<
                     Dataset<IN> input,
                     UnaryFunction<IN, KEY> keyExtractor,
                     UnaryFunction<IN, VALUE> valueExtractor,
-                    StateFactory<OUT, STATE> stateFactory,
+                    StateFactory<VALUE, OUT, STATE> stateFactory,
                     CombinableReduceFunction<STATE> stateCombiner,
                     @Nullable Windowing<WIN, W> windowing,
                     @Nullable ExtractEventTime<WIN> eventTimeAssigner,
@@ -253,7 +253,7 @@ public static OfBuilder named(String name) {
     return new OfBuilder(name);
   }
 
-  private final StateFactory<OUT, STATE> stateFactory;
+  private final StateFactory<VALUE, OUT, STATE> stateFactory;
   private final UnaryFunction<KIN, VALUE> valueExtractor;
   private final CombinableReduceFunction<STATE> stateCombiner;
 
@@ -264,7 +264,7 @@ public static OfBuilder named(String name) {
                    UnaryFunction<KIN, VALUE> valueExtractor,
                    @Nullable Windowing<WIN, W> windowing,
                    @Nullable ExtractEventTime<WIN> eventTimeAssigner,
-                   StateFactory<OUT, STATE> stateFactory,
+                   StateFactory<VALUE, OUT, STATE> stateFactory,
                    CombinableReduceFunction<STATE> stateCombiner,
                    Partitioning<KEY> partitioning)
   {
@@ -274,7 +274,7 @@ public static OfBuilder named(String name) {
     this.stateCombiner = stateCombiner;
   }
 
-  public StateFactory<OUT, STATE> getStateFactory() {
+  public StateFactory<VALUE, OUT, STATE> getStateFactory() {
     return stateFactory;
   }
 
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/TopPerKey.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/TopPerKey.java
index 88a474ddf048d..0514be3b11bcf 100644
--- a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/TopPerKey.java
+++ b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/TopPerKey.java
@@ -26,6 +26,7 @@
 import cz.seznam.euphoria.core.client.graph.DAG;
 import cz.seznam.euphoria.core.client.io.Context;
 import cz.seznam.euphoria.core.client.operator.state.State;
+import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.operator.state.StorageProvider;
 import cz.seznam.euphoria.core.client.operator.state.ValueStorage;
 import cz.seznam.euphoria.core.client.operator.state.ValueStorageDescriptor;
@@ -294,13 +295,13 @@ public UnaryFunction<IN, SCORE> getScoreExtractor() {
         MaxScored<VALUE, SCORE>, W>
         reduce =
         new ReduceStateByKey<>(getName() + "::ReduceStateByKey", flow, input,
-            keyExtractor,
-            e -> Pair.of(valueFn.apply(e), scoreFn.apply(e)),
-            windowing,
-            eventTimeAssigner,
-            MaxScored::new,
-            stateCombiner,
-            partitioning);
+                keyExtractor,
+                e -> Pair.of(valueFn.apply(e), scoreFn.apply(e)),
+                windowing,
+                eventTimeAssigner,
+                (StateFactory<Pair<VALUE, SCORE>, Pair<VALUE, SCORE>, MaxScored<VALUE, SCORE>>) MaxScored::new,
+                stateCombiner,
+                partitioning);
 
     MapElements<Pair<KEY, Pair<VALUE, SCORE>>, Triple<KEY, VALUE, SCORE>>
         format =
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/state/StateFactory.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/state/StateFactory.java
index e4c770bbd29f9..51cb626ca5a75 100644
--- a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/state/StateFactory.java
+++ b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/state/StateFactory.java
@@ -15,14 +15,16 @@
  */
 package cz.seznam.euphoria.core.client.operator.state;
 
-import cz.seznam.euphoria.core.client.functional.BinaryFunction;
 import cz.seznam.euphoria.core.client.io.Context;
-import cz.seznam.euphoria.core.client.operator.state.StorageProvider;
+
+import java.io.Serializable;
 
 /**
  * Factory for states.
  */
-public interface StateFactory<T, STATE>
-    extends BinaryFunction<Context<T>, StorageProvider, STATE> {
+@FunctionalInterface
+public interface StateFactory<IN, OUT, STATE extends State<IN, OUT>> extends Serializable {
+
+  STATE createState(Context<OUT> context, StorageProvider storageProvider);
 
 }
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/executor/greduce/GroupReducer.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/executor/greduce/GroupReducer.java
index 05c45679dd561..f587e66aad5f3 100644
--- a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/executor/greduce/GroupReducer.java
+++ b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/executor/greduce/GroupReducer.java
@@ -22,12 +22,12 @@
 import cz.seznam.euphoria.core.client.dataset.windowing.Windowing;
 import cz.seznam.euphoria.core.client.functional.BinaryFunction;
 import cz.seznam.euphoria.core.client.functional.CombinableReduceFunction;
-import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.io.Context;
 import cz.seznam.euphoria.core.client.operator.state.ListStorage;
 import cz.seznam.euphoria.core.client.operator.state.ListStorageDescriptor;
 import cz.seznam.euphoria.core.client.operator.state.MergingStorageDescriptor;
 import cz.seznam.euphoria.core.client.operator.state.State;
+import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.operator.state.Storage;
 import cz.seznam.euphoria.core.client.operator.state.StorageDescriptor;
 import cz.seznam.euphoria.core.client.operator.state.StorageProvider;
@@ -68,7 +68,7 @@ public interface WindowedElementFactory<W extends Window, T> {
     WindowedElement<W, T> create(W window, long timestamp, T element);
   }
 
-  private final StateFactory<?, State> stateFactory;
+  private final StateFactory<I, ?, State<I, ?>> stateFactory;
   private final WindowedElementFactory<WID, Object> elementFactory;
   private final CombinableReduceFunction<State> stateCombiner;
   private final StorageProvider stateStorageProvider;
@@ -82,7 +82,7 @@ public interface WindowedElementFactory<W extends Window, T> {
   final HashMap<WID, State> states = new HashMap<>();
   KEY key;
 
-  public GroupReducer(StateFactory<?, State> stateFactory,
+  public GroupReducer(StateFactory<I, ?, State<I, ?>> stateFactory,
                       WindowedElementFactory<WID, Object> elementFactory,
                       CombinableReduceFunction<State> stateCombiner,
                       StorageProvider stateStorageProvider,
@@ -123,7 +123,7 @@ public void process(WindowedElement<WID, Pair<KEY, I>> elem) {
     {
       State state = states.get(window);
       if (state == null) {
-        state = stateFactory.apply(
+        state = stateFactory.createState(
             new ElementCollectContext(collector, window), stateStorageProvider);
         states.put(window, state);
       }
diff --git a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/batch/ReduceStateByKeyTranslator.java b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/batch/ReduceStateByKeyTranslator.java
index 287bd549b4a14..35684c301535a 100644
--- a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/batch/ReduceStateByKeyTranslator.java
+++ b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/batch/ReduceStateByKeyTranslator.java
@@ -128,8 +128,8 @@ static class RSBKReducer
           implements GroupReduceFunction<BatchElement<?, Pair>, BatchElement<?, Pair>>,
           ResultTypeQueryable<BatchElement<?, Pair>>
   {
-    private final StateFactory<?, State> stateFactory;
-    private final CombinableReduceFunction<State> stateCombiner;
+    private final StateFactory<?, ?, State<?, ?>> stateFactory;
+    private final CombinableReduceFunction<State<?, ?>> stateCombiner;
     private final StorageProvider stateStorageProvider;
     private final Windowing windowing;
     private final Trigger trigger;
@@ -152,7 +152,7 @@ static class RSBKReducer
     public void reduce(Iterable<BatchElement<?, Pair>> values,
                        org.apache.flink.util.Collector<BatchElement<?, Pair>> out)
     {
-      GroupReducer reducer = new GroupReducer<>(
+      GroupReducer reducer = new GroupReducer(
               stateFactory,
               BatchElement::new,
               stateCombiner,
diff --git a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java
index 5ea33ca2d46bb..f414d9f61dbc8 100644
--- a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java
+++ b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/ReduceStateByKeyTranslator.java
@@ -73,8 +73,8 @@ public DataStream<?> translate(FlinkOperator<ReduceStateByKey> operator,
 
     ReduceStateByKey origOperator = operator.getOriginalOperator();
 
-    StateFactory<?, State> stateFactory = origOperator.getStateFactory();
-    CombinableReduceFunction stateCombiner = origOperator.getStateCombiner();
+    StateFactory<?, ?, State<?, ?>> stateFactory = origOperator.getStateFactory();
+    CombinableReduceFunction<State<?, ?>> stateCombiner = origOperator.getStateCombiner();
 
     Windowing windowing = origOperator.getWindowing();
     if (windowing == null) {
diff --git a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/AbstractWindowOperator.java b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/AbstractWindowOperator.java
index a188cddb64f3b..ee0ff77976bd3 100644
--- a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/AbstractWindowOperator.java
+++ b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/AbstractWindowOperator.java
@@ -20,12 +20,12 @@
 import cz.seznam.euphoria.core.client.dataset.windowing.Window;
 import cz.seznam.euphoria.core.client.dataset.windowing.Windowing;
 import cz.seznam.euphoria.core.client.functional.CombinableReduceFunction;
-import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.io.Context;
 import cz.seznam.euphoria.core.client.operator.state.ListStorage;
 import cz.seznam.euphoria.core.client.operator.state.ListStorageDescriptor;
 import cz.seznam.euphoria.core.client.operator.state.MergingStorageDescriptor;
 import cz.seznam.euphoria.core.client.operator.state.State;
+import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.operator.state.StorageDescriptor;
 import cz.seznam.euphoria.core.client.operator.state.ValueStorage;
 import cz.seznam.euphoria.core.client.operator.state.ValueStorageDescriptor;
@@ -65,8 +65,8 @@ public abstract class AbstractWindowOperator<I, KEY, WID extends Window>
 
   private final Windowing<?, WID> windowing;
   private final Trigger<WID> trigger;
-  private final StateFactory<?, State> stateFactory;
-  private final CombinableReduceFunction<State> stateCombiner;
+  private final StateFactory<?, ?, State<?, ?>> stateFactory;
+  private final CombinableReduceFunction<State<?, ?>> stateCombiner;
 
 
   // FIXME Arguable hack that ensures all remaining opened windows
@@ -94,8 +94,8 @@ public abstract class AbstractWindowOperator<I, KEY, WID extends Window>
   private transient TypeSerializer<WID> windowSerializer;
 
   public AbstractWindowOperator(Windowing<?, WID> windowing,
-                                StateFactory<?, State> stateFactory,
-                                CombinableReduceFunction<State> stateCombiner,
+                                StateFactory<?, ?, State<?, ?>> stateFactory,
+                                CombinableReduceFunction<State<?, ?>> stateCombiner,
                                 boolean localMode,
                                 int descriptorsCacheMaxSize) {
     this.windowing = Objects.requireNonNull(windowing);
@@ -191,7 +191,7 @@ public void processElement(StreamRecord<I> record)
                   List<State> states = new ArrayList<>();
                   states.add(getWindowState(stateResultWindow));
                   mergedStateWindows.forEach(sw -> states.add(getWindowState(sw)));
-                  stateCombiner.apply(states);
+                  stateCombiner.apply((Iterable) states);
 
                   // remove merged window states
                   mergedStateWindows.forEach(sw -> {
@@ -322,7 +322,7 @@ private void processTriggerResult(WID window,
   @SuppressWarnings("unchecked")
   private State getWindowState(WID window) {
     storageProvider.setWindow(window);
-    return stateFactory.apply(outputContext, storageProvider);
+    return stateFactory.createState(outputContext, storageProvider);
   }
 
   private MergingWindowSet<WID> getMergingWindowSet()  {
diff --git a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/KeyedMultiWindowedElementWindowOperator.java b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/KeyedMultiWindowedElementWindowOperator.java
index d73608f7e9291..e6dded5a58346 100644
--- a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/KeyedMultiWindowedElementWindowOperator.java
+++ b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/KeyedMultiWindowedElementWindowOperator.java
@@ -31,8 +31,8 @@ public class KeyedMultiWindowedElementWindowOperator<KEY, WID extends Window>
 
     public KeyedMultiWindowedElementWindowOperator(
             Windowing<?, WID> windowing,
-            StateFactory<?, State> stateFactory,
-            CombinableReduceFunction<State> stateCombiner,
+            StateFactory<?, ?, State<?, ?>> stateFactory,
+            CombinableReduceFunction<State<?, ?>> stateCombiner,
             boolean localMode,
             int descriptorsCacheMaxSize) {
         super(windowing, stateFactory, stateCombiner, localMode, descriptorsCacheMaxSize);
diff --git a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/StreamingElementWindowOperator.java b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/StreamingElementWindowOperator.java
index 83c99b39692cb..4512fc7f807c1 100644
--- a/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/StreamingElementWindowOperator.java
+++ b/sdks/java/extensions/euphoria/euphoria-flink/src/main/java/cz/seznam/euphoria/flink/streaming/windowing/StreamingElementWindowOperator.java
@@ -38,8 +38,8 @@ public class StreamingElementWindowOperator<KEY, WID extends Window>
   public StreamingElementWindowOperator(
           WindowAssigner<?, KEY, ?, WID> windowAssigner,
           Windowing<?, WID> windowing,
-          StateFactory<?, State> stateFactory,
-          CombinableReduceFunction<State> stateCombiner,
+          StateFactory<?, ?, State<?, ?>> stateFactory,
+          CombinableReduceFunction<State<?, ?>> stateCombiner,
           boolean localMode,
           int descriptorsCacheMaxSize) {
     super(windowing, stateFactory, stateCombiner, localMode, descriptorsCacheMaxSize);
diff --git a/sdks/java/extensions/euphoria/euphoria-flink/src/test/java/cz/seznam/euphoria/flink/streaming/RSBKWindowingTest.java b/sdks/java/extensions/euphoria/euphoria-flink/src/test/java/cz/seznam/euphoria/flink/streaming/RSBKWindowingTest.java
index a425284b3b0b4..4b4ce3e2c980e 100644
--- a/sdks/java/extensions/euphoria/euphoria-flink/src/test/java/cz/seznam/euphoria/flink/streaming/RSBKWindowingTest.java
+++ b/sdks/java/extensions/euphoria/euphoria-flink/src/test/java/cz/seznam/euphoria/flink/streaming/RSBKWindowingTest.java
@@ -111,8 +111,7 @@ public void testEventWindowing() throws Exception {
         ReduceStateByKey.of(f.createInput(source))
         .keyBy(Pair::getFirst)
         .valueBy(e -> e)
-        .stateFactory((StateFactory<Pair<String, Integer>, AccState<Pair<String,
-            Integer>>>) AccState::new)
+        .stateFactory((StateFactory<Pair<String, Integer>, Pair<String, Integer>, AccState<Pair<String, Integer>>>) AccState::new)
         .combineStateBy(AccState::combine)
         .windowBy(Time.of(Duration.ofMillis(5)),
             // ~ event time
@@ -163,7 +162,7 @@ public void testEventWindowing_attachedWindowing() throws Exception {
         ReduceStateByKey.of(f.createInput(source))
             .keyBy(Pair::getFirst)
             .valueBy(e -> e)
-            .stateFactory((StateFactory<Pair<String, Integer>, AccState<Pair<String, Integer>>>) AccState::new)
+            .stateFactory((StateFactory<Pair<String, Integer>, Pair<String, Integer>, AccState<Pair<String, Integer>>>) AccState::new)
             .combineStateBy(AccState::combine)
             .windowBy(Time.of(Duration.ofMillis(5)),
                 // ~ event time
@@ -178,8 +177,7 @@ public void testEventWindowing_attachedWindowing() throws Exception {
         ReduceStateByKey.of(secondStep)
         .keyBy(Pair::getFirst)
         .valueBy(e -> e)
-        .stateFactory((StateFactory<Pair<String, Integer>, AccState<Pair<String,
-            Integer>>>) AccState::new)
+        .stateFactory((StateFactory<Pair<String, Integer>, Pair<String, Integer>, AccState<Pair<String, Integer>>>) AccState::new)
         .combineStateBy(AccState::combine)
         .windowBy(Time.of(Duration.ofMillis(5)),
             // ~ event time
diff --git a/sdks/java/extensions/euphoria/euphoria-inmem/src/main/java/cz/seznam/euphoria/inmem/ReduceStateByKeyReducer.java b/sdks/java/extensions/euphoria/euphoria-inmem/src/main/java/cz/seznam/euphoria/inmem/ReduceStateByKeyReducer.java
index 936cb6a0a7575..4447e9644b9c9 100644
--- a/sdks/java/extensions/euphoria/euphoria-inmem/src/main/java/cz/seznam/euphoria/inmem/ReduceStateByKeyReducer.java
+++ b/sdks/java/extensions/euphoria/euphoria-inmem/src/main/java/cz/seznam/euphoria/inmem/ReduceStateByKeyReducer.java
@@ -398,7 +398,7 @@ final class ProcessingState {
     final Collector<Datum> stateOutput;
     final BlockingQueue<Datum> rawOutput;
     final TriggerScheduler<Window, Object> triggering;
-    final StateFactory<Object, State> stateFactory;
+    final StateFactory<Object, Object, State<Object, Object>> stateFactory;
     final CombinableReduceFunction<State> stateCombiner;
 
     final ProcessingStats stats = new ProcessingStats(this);
@@ -409,7 +409,7 @@ final class ProcessingState {
     private ProcessingState(
         BlockingQueue<Datum> output,
         TriggerScheduler<Window, Object> triggering,
-        StateFactory<Object, State> stateFactory,
+        StateFactory<Object, Object, State<Object, Object>> stateFactory,
         CombinableReduceFunction<State> stateCombiner,
         StorageProvider storageProvider) {
 
@@ -484,7 +484,7 @@ State getWindowStateForUpdate(KeyedWindow<?, ?> kw) {
       State state = wRegistry.getWindowState(kw);
       if (state == null) {
         // ~ if no such window yet ... set it up
-        state = stateFactory.apply(
+        state = stateFactory.createState(
                 new KeyedElementCollector(
                     stateOutput, kw.window(), kw.key(),
                     processing.triggering::getCurrentTimestamp),
diff --git a/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceStateByKeyTest.java b/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceStateByKeyTest.java
index a5f8223c9d8cc..f0ec1317cf16b 100644
--- a/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceStateByKeyTest.java
+++ b/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceStateByKeyTest.java
@@ -302,7 +302,7 @@ protected Partitions<Pair<String, Integer>> getInput() {
         return ReduceStateByKey.of(input)
             .keyBy(e -> e.getFirst().charAt(0) - '0')
             .valueBy(Pair::getFirst)
-            .stateFactory((StateFactory<Long, CountState>) CountState::new)
+            .stateFactory((StateFactory<String, Long, CountState>) CountState::new)
             .combineStateBy(CountState::combine)
             // FIXME .timedBy(Pair::getSecond) and make the assertion in the validation phase stronger
             .windowBy(Count.of(3))
@@ -405,7 +405,7 @@ protected Partitions<Pair<String, Integer>> getInput() {
             ReduceStateByKey.of(input)
                 .keyBy(e -> e.getFirst().charAt(0) - '0')
                 .valueBy(Pair::getFirst)
-                .stateFactory((StateFactory<String, AccState<String>>) AccState::new)
+                .stateFactory((StateFactory<String, String, AccState<String>>) AccState::new)
                 .combineStateBy(AccState::combine)
                 .windowBy(Time.of(Duration.ofSeconds(5)),
                         (Pair<String, Integer> e) -> e.getSecond() * 1_000L)
@@ -466,7 +466,7 @@ protected Dataset<Triple<TimeInterval, Integer, String>> getOutput(Dataset<Pair<
             ReduceStateByKey.of(input)
                 .keyBy(e -> e.getFirst().charAt(0) - '0')
                 .valueBy(e -> e.getFirst().substring(2))
-                .stateFactory((StateFactory<String, AccState<String>>) AccState::new)
+                .stateFactory((StateFactory<String, String, AccState<String>>) AccState::new)
                 .combineStateBy(AccState::combine)
                 .windowBy(TimeSliding.of(Duration.ofSeconds(10), Duration.ofSeconds(5)),
                         (Pair<String, Integer> e) -> e.getSecond() * 1_000L)
@@ -545,7 +545,7 @@ protected Partitions<Pair<String, Integer>> getInput() {
             ReduceStateByKey.of(input)
                 .keyBy(e -> e.getFirst().charAt(0) - '0')
                 .valueBy(Pair::getFirst)
-                .stateFactory((StateFactory<String, AccState<String>>) AccState::new)
+                .stateFactory((StateFactory<String, String, AccState<String>>) AccState::new)
                 .combineStateBy(AccState::combine)
                 .windowBy(Session.of(Duration.ofSeconds(5)),
                         (Pair<String, Integer> e) -> e.getSecond() * 1_000L)
diff --git a/sdks/java/extensions/euphoria/euphoria-spark/src/main/java/cz/seznam/euphoria/spark/ReduceStateByKeyTranslator.java b/sdks/java/extensions/euphoria/euphoria-spark/src/main/java/cz/seznam/euphoria/spark/ReduceStateByKeyTranslator.java
index a4457c6fc3231..b4454d897d68a 100644
--- a/sdks/java/extensions/euphoria/euphoria-spark/src/main/java/cz/seznam/euphoria/spark/ReduceStateByKeyTranslator.java
+++ b/sdks/java/extensions/euphoria/euphoria-spark/src/main/java/cz/seznam/euphoria/spark/ReduceStateByKeyTranslator.java
@@ -19,11 +19,11 @@
 import cz.seznam.euphoria.core.client.dataset.windowing.Window;
 import cz.seznam.euphoria.core.client.dataset.windowing.Windowing;
 import cz.seznam.euphoria.core.client.functional.CombinableReduceFunction;
-import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.functional.UnaryFunction;
 import cz.seznam.euphoria.core.client.operator.ExtractEventTime;
 import cz.seznam.euphoria.core.client.operator.ReduceStateByKey;
 import cz.seznam.euphoria.core.client.operator.state.State;
+import cz.seznam.euphoria.core.client.operator.state.StateFactory;
 import cz.seznam.euphoria.core.client.triggers.Trigger;
 import cz.seznam.euphoria.core.client.util.Pair;
 import cz.seznam.euphoria.core.executor.greduce.GroupReducer;
@@ -53,8 +53,8 @@ public JavaRDD<?> translate(ReduceStateByKey operator,
 
     final JavaRDD<SparkElement> input = (JavaRDD) context.getSingleInput(operator);
 
-    StateFactory<?, State> stateFactory = operator.getStateFactory();
-    CombinableReduceFunction<State> stateCombiner = operator.getStateCombiner();
+    StateFactory<?, ?, State<?, ?>> stateFactory = operator.getStateFactory();
+    CombinableReduceFunction<State<?, ?>> stateCombiner = operator.getStateCombiner();
 
     final UnaryFunction keyExtractor = operator.getKeyExtractor();
     final UnaryFunction valueExtractor = operator.getValueExtractor();
@@ -155,16 +155,16 @@ private static class StateReducer
 
     private final Windowing windowing;
     private final Trigger trigger;
-    private final StateFactory<?, State> stateFactory;
-    private final CombinableReduceFunction<State> stateCombiner;
+    private final StateFactory<?, ?, State<?, ?>> stateFactory;
+    private final CombinableReduceFunction<State<?, ?>> stateCombiner;
     private final SparkStorageProvider storageProvider;
 
     // mapping of [Key -> GroupReducer]
     private transient Map<Object, GroupReducer> activeReducers;
 
     public StateReducer(Windowing windowing,
-                        StateFactory<?, State> stateFactory,
-                        CombinableReduceFunction<State> stateCombiner) {
+                        StateFactory<?, ?, State<?, ?>> stateFactory,
+                        CombinableReduceFunction<State<?, ?>> stateCombiner) {
       this.windowing = windowing;
       this.trigger = windowing.getTrigger();
       this.stateFactory = stateFactory;
@@ -195,7 +195,7 @@ public Iterator<SparkElement> call(Iterator<Tuple2<KeyedWindow, Object>> iterato
 
           GroupReducer reducer = activeReducers.get(kw.key());
           if (reducer == null) {
-            reducer = new GroupReducer<>(stateFactory,
+            reducer = new GroupReducer(stateFactory,
                     SparkElement::new,
                     stateCombiner,
                     storageProvider,