The following implementation is a hybrid model, where the push model is going to call the pull model causing reentrancy
484521
private boolean updateMemoryReservation()
    {
        // Operator/driver will be blocked on memory after we call localUserMemoryContext.setBytes().
        // If memory is not available, once we return, this operator will be blocked until memory is available.
        long memorySizeInBytes = groupByHash.map(GroupByHash::getEstimatedSize).orElse(0L) + partitionRowCount.sizeOf();
        localUserMemoryContext.setBytes(memorySizeInBytes);
        // If memory is not available, inform the caller that we cannot proceed for allocation.
        return operatorContext.isWaitingForMemory().isDone();
    }
1085547
function handlerEnteredOrUpdated(currentHandlerInfos, handlerInfo, enter, transition) {

    var handler = handlerInfo.handler,
        context = handlerInfo.context;

    if (enter) {
      _routerUtils.callHook(handler, 'enter', transition);
    }
    if (transition && transition.isAborted) {
      throw new _routerTransition.TransitionAborted();
    }

    handler.context = context;
    _routerUtils.callHook(handler, 'contextDidChange');

    _routerUtils.callHook(handler, 'setup', context, transition);
    if (transition && transition.isAborted) {
      throw new _routerTransition.TransitionAborted();
    }

    currentHandlerInfos.push(handlerInfo);

    return true;
  }
621599
private <IN1, IN2, OUT> Collection<Integer> transformTwoInputTransform(TwoInputTransformation<IN1, IN2, OUT> transform) {

		Collection<Integer> inputIds1 = transform(transform.getInput1());
		Collection<Integer> inputIds2 = transform(transform.getInput2());

		// the recursive call might have already transformed this
		if (alreadyTransformed.containsKey(transform)) {
			return alreadyTransformed.get(transform);
		}

		List<Integer> allInputIds = new ArrayList<>();
		allInputIds.addAll(inputIds1);
		allInputIds.addAll(inputIds2);

		String slotSharingGroup = determineSlotSharingGroup(transform.getSlotSharingGroup(), allInputIds);

		streamGraph.addCoOperator(
				transform.getId(),
				slotSharingGroup,
				transform.getCoLocationGroupKey(),
				transform.getOperator(),
				transform.getInputType1(),
				transform.getInputType2(),
				transform.getOutputType(),
				transform.getName());

		if (transform.getStateKeySelector1() != null || transform.getStateKeySelector2() != null) {
			TypeSerializer<?> keySerializer = transform.getStateKeyType().createSerializer(env.getConfig());
			streamGraph.setTwoInputStateKey(transform.getId(), transform.getStateKeySelector1(), transform.getStateKeySelector2(), keySerializer);
		}

		streamGraph.setParallelism(transform.getId(), transform.getParallelism());
		streamGraph.setMaxParallelism(transform.getId(), transform.getMaxParallelism());

		for (Integer inputId: inputIds1) {
			streamGraph.addEdge(inputId,
					transform.getId(),
					1
			);
		}

		for (Integer inputId: inputIds2) {
			streamGraph.addEdge(inputId,
					transform.getId(),
					2
			);
		}

		return Collections.singleton(transform.getId());
	}
988625
function joinCurrentModelActivity(method, stateName, modelOrCollection, context, force) {
        var xhrActivity = modelOrCollection[xhrModelLoadingAttribute];
        if (xhrActivity) {
            _.each(xhrActivity, function(xhrEvent) {
                if (!method || method === ALL_XHR_ACTIVITY || xhrEvent.method === method) {
                    // this is one that is applicable
                    pushLoadingState(xhrEvent, stateName, modelOrCollection, context, force);
                }
            });
        }
    }
224278
@Override
	public void computeUnclosedBranchStack() {
		if (this.openBranches != null) {
			return;
		}

		addClosedBranches(getPredecessorNode().closedBranchingNodes);
		List<UnclosedBranchDescriptor> fromInput = getPredecessorNode().getBranchesForParent(this.inConn);
		
		// handle the data flow branching for the broadcast inputs
		List<UnclosedBranchDescriptor> result = computeUnclosedBranchStackForBroadcastInputs(fromInput);
		
		this.openBranches = (result == null || result.isEmpty()) ? Collections.<UnclosedBranchDescriptor>emptyList() : result;
	}
488988
private void performPersisting(ConvertedCommit commit, EKBCommit source) throws EKBException {
        try {
            EDBCommit ci = edbService.createEDBCommit(commit.getInserts(), commit.getUpdates(), commit.getDeletes());
            ci.setDomainId(source.getDomainId());
            ci.setConnectorId(source.getConnectorId());
            ci.setInstanceId(source.getInstanceId());
            ci.setComment(source.getComment());
            edbService.commit(ci);
            source.setRevisionNumber(ci.getRevisionNumber());
            source.setParentRevisionNumber(ci.getParentRevisionNumber());
        } catch (EDBCheckException e) {
            throw new ModelPersistException(convertEDBObjectList(e.getFailedInserts()),
                convertEDBObjectList(e.getFailedUpdates()), e.getFailedDeletes(), e);
        } catch (EDBException e) {
            throw new EKBException("Error while commiting EKBCommit", e);
        }
    }
16234
final void bipush(CompletableFuture<?> b, BiCompletion<?, ?, ?> c) {
    if (c != null) {
      while (result == null) {
        if (tryPushStack(c)) {
          if (b.result == null)
            b.unipush(new CoCompletion(c));
          else if (result != null)
            c.tryFire(SYNC);
          return;
        }
      }
      b.unipush(c);
    }
  }
103877
final void cmdRemoveExpiring(final long lockId, final PersistentTransaction transaction) throws ProtocolException, TransactionException, SevereMessageStoreException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "cmdRemoveExpiring", new Object[] { "Item Link:   " + this, "Stream Link: " + _owningStreamLink, formatLockId(lockId),
                                                                     "Transaction: " + transaction });

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(this, tc, "expiring item " + _tuple.getUniqueId());

        synchronized (this)
        {
            if (ItemLinkState.STATE_LOCKED_FOR_EXPIRY == _itemLinkState)
            {
                if (_lockID != lockId)
                {
                    throw new LockIdMismatch(_lockID, lockId);
                }

                ListStatistics stats = getParentStatistics();
                synchronized (stats)
                {
                    stats.decrementExpiring();
                    stats.incrementRemoving();
                }

                _transactionId = transaction.getPersistentTranId();
                _itemLinkState = ItemLinkState.STATE_REMOVING_EXPIRING;
            }
            else
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled())
                    SibTr.event(this, tc, "Invalid Item state: " + _itemLinkState);
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(this, tc, "cmdRemoveExpiring");
                throw new StateException(_itemLinkState.toString());
            }
        }
        final Task task = new RemoveLockedTask(this);
        transaction.addWork(task);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "cmdRemoveExpiring");
    }
72310
<T> T callWithRetries(Callable<T> callable, RecurringPolicy<?> recurringPolicy,
      RecurringStats retryStats, Set<Class<? extends Exception>> retryableExceptions,
      boolean recoverable, boolean logFailures) throws Exception {
    boolean recovery = retryStats != null;

    while (true) {
      try {
        return callable.call();
      } catch (Exception e) {
        ShutdownSignalException sse = extractCause(e, ShutdownSignalException.class);
        if (sse == null && logFailures && recurringPolicy != null
            && recurringPolicy.allowsAttempts())
          log.error("Invocation of {} failed.", callable, e);

        if (sse != null && (recovery || !recoverable))
          throw e;

        if (!closed) {
          try {
            // Retry on channel recovery failure or retryable exception
            boolean retryable =
                recurringPolicy != null && recurringPolicy.allowsAttempts()
                    && isRetryable(retryableExceptions, e, sse);
            long startTime = System.nanoTime();

            if (retryable) {
              if (retryStats == null) 
                retryStats = new RecurringStats(recurringPolicy);
                
              // Wait for pending recovery
              if (sse != null) {
                if (recurringPolicy.getMaxDuration() == null)
                  circuit.await();
                else if (!circuit.await(retryStats.getMaxWaitTime())) {
                  log.debug("Exceeded max wait time while waiting for {} to recover", this);
                  throw e;
                }
              }

              // Continue retries
              retryStats.incrementAttempts();
              if (!retryStats.isPolicyExceeded()) {
                long remainingWaitTime =
                    retryStats.getWaitTime().toNanos() - (System.nanoTime() - startTime);
                if (remainingWaitTime > 0)
                  retryWaiter.await(Duration.nanos(remainingWaitTime));
                continue;
              }
            }
          } catch (Throwable ignore) {
          }
        }

        throw e;
      }
    }
  }
1150983
protected synchronized void switchEntry(boolean blocking) throws IOException {
    /**
     * Must create compaction redo before update redo to prevent data loss.
     * 
     * The "applyEntries" method will internally sort compaction redo and update redo
     * entries using stable Arrays.sort. The following order of creating redo entries
     * will ensure that compaction redo does NOT overwrite update redo.
     */
    if (!_entryCompaction.isEmpty()) {
      // Create entry log and persist in-memory data
      File file = new File(getDirectory(), getEntryLogName(_entryCompaction));
      _entryCompaction.save(file);
      _entryPool.addToServiceQueue(_entryCompaction);
      _entryCompaction = _entryPool.next();
    }
    
    if (!_entry.isEmpty()) {
      if(_persistListener != null) {
        _persistListener.beforePersist(_entry);
      }
      
      // Create entry log and persist in-memory data
      File file = new File(getDirectory(), getEntryLogName(_entry));
      _entry.save(file);
      
      if(_persistListener != null) {
        _persistListener.afterPersist(_entry);
      }
      
      // Advance low water mark to maintain progress record
      _lwmScn = Math.max(_lwmScn, _entry.getMaxScn());
      _entryPool.addToServiceQueue(_entry);
      _entry = _entryPool.next();
    }
    
    // Apply entry logs to array file
    if (_autoApplyEntries) {
      if (_entryPool.getServiceQueueSize() >= _maxEntries) {
        applyEntries(blocking);
      }
    }
  }
868227
@Override
    public void postInvoke(int id, EJSDeployedSupport s) throws RemoteException {
        final boolean isTraceOn = TraceComponent.isAnyTracingEnabled();
        if (isTraceOn && tc.isEntryEnabled()) {
            Tr.entry(tc, "postInvoke");
        }

        // Release lock acquired if container managed concurrency control.
        if (ivContainerManagedConcurrency && s.ivLockAcquired) {
            // Get the lock type to use for the method being invoked.
            EJBMethodInfoImpl mInfo = s.methodInfo;
            LockType lockType = mInfo.ivLockType;

            if (lockType == LockType.READ) {
                ivReadLock.unlock();

                if (isTraceOn && tc.isDebugEnabled()) {
                    Tr.debug(tc, "postInvoke released read lock: " + ivLock.toString());
                }
            } else {
                ivWriteLock.unlock();

                if (isTraceOn && tc.isDebugEnabled()) {
                    Tr.debug(tc, "postInvoke released write lock: " + ivLock.toString());
                }
            }
        }

        if (isTraceOn && tc.isEntryEnabled()) {
            Tr.exit(tc, "postInvoke");
        }
    }
285251
function transformWithCancel(cancel) {
  let pulled = false;
  let backpressureChangePromiseResolve;
  let outputController;
  return {
    readable: new ReadableStream({
      start(controller) {
        outputController = controller;
      },
      pull() {
        if (backpressureChangePromiseResolve) {
          backpressureChangePromiseResolve();
        } else {
          pulled = true;
        }
      },
      cancel
    }, {highWaterMark: 0}),
    writable: new WritableStream({
      write: async function(chunk) {
        outputController.enqueue(chunk);
        if (!pulled) {
          await new Promise(resolve => {
            backpressureChangePromiseResolve = resolve;
          });
          backpressureChangePromiseResolve = null;
        } else {
          pulled = false;
        }
      },
      close: outputController.close.bind(outputController),
      abort: outputController.error.bind(outputController)
    })
  };
}
427337
private void performInsertEOLogic(EngineeringObjectModelWrapper model) {
        for (Field field : model.getForeignKeyFields()) {
            mergeEngineeringObjectWithReferencedModel(field, model);
        }
    }
61766
private void pushState(Node<K,V>[] t, int i, int n) {
            TableStack<K,V> s = spare;  // reuse if possible
            if (s != null)
                spare = s.next;
            else
                s = new TableStack<K,V>();
            s.tab = t;
            s.length = n;
            s.index = i;
            s.next = stack;
            stack = s;
        }
659676
@Override
	public void propagate(int evtmask) throws ContradictionException {
		if (waitFirstSol && model.getSolver().getSolutionCount() == 0) {
			return;//the UB does not allow to prune
		}
		// initialisation
		mandatoryArcsList.clear();
		for (int i = 0; i < n; i++) {
			Dmin[i] = D[i].getLB();
			Dmax[i] = D[i].getUB();
		}
		for (int i = 0; i < n; i++) {
			ISet nei = gV.getMandNeighOf(i);
			for (int j : nei) {
				if (i < j) {
					mandatoryArcsList.add(i * n + j);
				}
			}
		}
		updateCosts();
		lagrangianRelaxation();
	}
1149182
@UiThread
    public void rebindIfNeeded() {
        ThreadUtils.ensureMainThread();

        Iterator<RouterTransaction> backstackIterator = backstack.reverseIterator();
        while (backstackIterator.hasNext()) {
            RouterTransaction transaction = backstackIterator.next();

            if (transaction.controller.getNeedsAttach()) {
                performControllerChange(transaction, null, true, new SimpleSwapChangeHandler(false));
            } else {
                setControllerRouter(transaction.controller);
            }
        }
    }
366520
private ParallelTransition buildSlideTransition(final boolean isReverse, final SlideModel<SlideStep> previousSlideModel, final SlideModel<SlideStep> selectedSlideModel) {
        final ParallelTransition slideAnimation = ParallelTransitionBuilder.create().build();

        if (previousSlideModel != null) {
            final Animation a = isReverse ? previousSlideModel.getShowAnimation() : previousSlideModel.getHideAnimation();
            if (a != null) {
                slideAnimation.getChildren().add(a);
            }
        }
        if (this.selectedSlideModel != null) {
            final Animation a = isReverse ? this.selectedSlideModel.getHideAnimation() : this.selectedSlideModel.getShowAnimation();
            if (a != null) {
                slideAnimation.getChildren().add(a);
            }
        }

        final SlideModel<SlideStep> csm = selectedSlideModel;
        // final SlideModel<SlideStep> psm = previousSlideModel;
        slideAnimation.setOnFinished(new javafx.event.EventHandler<ActionEvent>() {

            @Override
            public void handle(final ActionEvent arg0) {
                // // Hide previous slide
                // if (psm != null) {
                // getView().getRootNode().getChildren().removeAll(psm.getRootNode());
                // }

                // FIX ME NOT APPROPRIATED !!!
                if (previousSlideModel instanceof AbstractBaseModel) {
                    ((AbstractBaseModel<?>) previousSlideModel).doHideView(null);
                }
                if (selectedSlideModel instanceof AbstractBaseModel) {
                    ((AbstractBaseModel<?>) selectedSlideModel).doShowView(null);
                }

                // Hide all other slides
                if (csm != null) {
                    Node node;
                    for (int i = view().node().getChildren().size() - 1; i >= 0; i--) {
                        node = view().node().getChildren().get(i);
                        if (csm.node() != node) {
                            view().node().getChildren().remove(node);
                        }
                    }
                }
            }
        });
        return slideAnimation;
    }