@Override
    public void returnGroupByStates(Map states) {
        String partitionFlowId = SiddhiAppContext.getPartitionFlowId();
        for (Iterator<Map.Entry<String, State>> iterator =
             ((Set<Map.Entry<String, State>>) states.entrySet()).iterator();
             iterator.hasNext(); ) {
            Map.Entry<String, State> stateEntry = iterator.next();
            State state = stateEntry.getValue();
            if (state.activeUseCount == 0) {
                try {
                    if (state.canDestroy()) {
                        iterator.remove();
                    }
                } catch (Throwable t) {
                    log.error("Dropping partition state for partition key '" + partitionFlowId +
                            "' and the group by key '" + stateEntry.getKey() + "', due to error! " + t.getMessage(), t);
                    iterator.remove();
                }
            } else if (state.activeUseCount < 0) {
                throw new SiddhiAppRuntimeException("State active count has reached less then zero for partition key '"
                        + partitionFlowId + "' and the group by key '" + stateEntry.getKey() + "', current value is " +
                        state.activeUseCount);
            }

        }
        if (states.isEmpty()) {
            states.remove(partitionFlowId);
        }
    }
State holder for  partitioned use case
modules/siddhi-core/src/main/java/io/siddhi/core/util/snapshot/state/PartitionStateHolder.java