docs: private synchronized void onStateChange(PartitionId partitionId, PrimitiveState state) {
      states.put(partitionId, state);
      switch (state) {
        case CONNECTED:
          if (this.state != PrimitiveState.CONNECTED && !states.containsValue(PrimitiveState.SUSPENDED) && !states.containsValue(PrimitiveState.CLOSED)) {
            this.state = PrimitiveState.CONNECTED;
            stateChangeListeners.forEach(l -> l.accept(PrimitiveState.CONNECTED));
          }
          break;
        case SUSPENDED:
          if (this.state == PrimitiveState.CONNECTED) {
            this.state = PrimitiveState.SUSPENDED;
            stateChangeListeners.forEach(l -> l.accept(PrimitiveState.SUSPENDED));
          }
          break;
        case CLOSED:
          if (this.state != PrimitiveState.CLOSED) {
            this.state = PrimitiveState.CLOSED;
            stateChangeListeners.forEach(l -> l.accept(PrimitiveState.CLOSED));
          }
          break;
      }
    }