docs: private void processAck(ControlAck ack)
                      throws SIRollbackException, SIConnectionLostException, SIResourceException, SIErrorException
      {
  
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.entry(tc, "processAck");
  
          SIBUuid12 stream = ack.getGuaranteedStreamUUID();
  
          // This code iterates through all of the OutputHandlers
          // associated with this InputHandler, checking their ack prefixes
          // The ack prefix of the InputStream is updated if the minimum of the
          // ack prefixes of the OutputStreams is greater than the current
          // InputStream ack prefix.
          List indexList = null;
          long index = 0;
          MessageItemReference msgItemRef = null;
  
          // This is the maximum tick in a stream
          long min = Long.MAX_VALUE;
  
          final Iterator outputHandlers =
                          _destination.getAllPubSubOutputHandlers().values().iterator();
  
          PubSubOutputHandler outputHandler = null;
  
          // TODO: In IME case we have to check the Ack on the TargetStream too
          // if we have one
          // But watch out for race where this has not been created yet because
          // writing to TargetStream happens after sending msg to remoteMEs
          // so it's just possible we get Acks back before we've done it.
          while (outputHandlers.hasNext())
          {
              outputHandler = (PubSubOutputHandler) outputHandlers.next();
  
              min = outputHandler.checkAck(ack, min);
          } // end while
  
          // Determine whether we are an IME 
          if (!_sourceStreamManager.hasStream(stream))
          {
              min = _targetStreamManager.checkAck(ack, min);
          }
  
          try
          {
              // We need to ensure that we read the ackPrefix, check it 
              // against the outputStream ackPrefixes and update it in one
              // operation  
              synchronized (this)
              {
                  // First determine whether this Ack is for the sourceStream
                  // or one of the internalInputStreams
                  if (_sourceStreamManager.hasStream(stream))
                  {
                      // Update the SourceStream ackPrefix
                      indexList = _sourceStreamManager.processAck(ack, min);
                  }
                  else // internal input stream
                  {
                      // Update the SourceStream ackPrefix
                      indexList = _internalInputStreamManager.processAck(ack, min);
                  }
              } // End sync
  
              // If this is the sourceME and we updated the ackPrefix
              // then this list will contain messages to be deleted
              // Otherwise it will always be null 
              if (indexList != null)
              {
                  TransactionCommon tran = _sourceBatchHandler.registerInBatch();
                  // A marker to indicate how far through the method we get. 
                  boolean messageAddCall = false;
  
                  try
                  {
                      SourceStream batchListener = _sourceStreamManager.getBatchListener(ack);
                      TickRange tr = null;
                      for (int i = 0; i < indexList.size(); i++)
                      {
                          tr = (TickRange) indexList.get(i);
                          batchListener.addToBatchList(tr);
  
                          index = tr.itemStreamIndex;
                          try
                          {
                              msgItemRef =
                                              (MessageItemReference) _proxyReferenceStream
                                                              .findById(
                                                              index);
                              //it is possible that this is null in the case of an expired msg writing
                              //silence into the stream
                              if (msgItemRef != null)
                              {
                                  Transaction msTran = _messageProcessor.resolveAndEnlistMsgStoreTransaction(tran);
                                  msgItemRef.remove(msTran, msgItemRef.getLockID());
                              }
                          } catch (MessageStoreException e)
                          {
                              // MessageStoreException shouldn't occur so FFDC.
                              FFDCFilter.processException(
                                                          e,
                                                          "com.ibm.ws.sib.processor.impl.PubSubInputHandler.processAck",
                                                          "1:951:1.329.1.1",
                                                          this);
  
                              SibTr.exception(tc, e);
  
                              if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                                  SibTr.exit(tc, "processAck", e);
  
                              SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                          new Object[] {
                                                        "com.ibm.ws.sib.processor.impl.PubSubInputHandler",
                                                        "1:962:1.329.1.1",
                                                        e });
  
                              throw new SIResourceException(
                                              nls.getFormattedMessage(
                                                                      "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                                      new Object[] {
                                                                                    "com.ibm.ws.sib.processor.impl.PubSubInputHandler",
                                                                                    "1:970:1.329.1.1",
                                                                                    e },
                                                                      null),
                                              e);
                          }
                      }
                      // Indicate that the messagesAdded call has been made
                      messageAddCall = true;
                      _sourceBatchHandler.messagesAdded(indexList.size(), batchListener);
                  } finally
                  {
                      if (!messageAddCall)
                          try
                          {
                              _sourceBatchHandler.messagesAdded(0);
                          } catch (SIResourceException e)
                          {
                              // No FFDC code needed, This will allow for any exceptions that were thrown to
                              // be rethrown instead of overiding with a batch handler error.
                              if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled())
                                  SibTr.exception(tc, e);
                          }
                  }
              } // End of deleting messages at source
  
          } finally
          {
              // unlock as the getAllPubSubOutputHandlers locks it.
              _destination.unlockPubsubOutputHandlers();
  
              // Before we exit, let the source stream manager take care of any pending flushes
              _sourceStreamManager.attemptFlushIfNecessary();
  
              if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                  SibTr.exit(tc, "processAck");
          }
      }