docs: public void handleNotFlushedMessage(ControlNotFlushed cMsg)
  
    throws SIIncorrectCallException, SIErrorException, SIResourceException
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.entry(this, tc, "handleNotFlushedMessage", new Object[] { cMsg });
  
      SIBUuid12 streamID = cMsg.getGuaranteedStreamUUID();
      //Synchronize to resolve racing messages.
      synchronized (flushMap)
      {
        FlushQueryRecord entry = flushMap.get(streamID);
        // make sure the request ID matches.
        long reqId = cMsg.getRequestID();
        //if there is no entry or the entry's request ID does not match then
        //it must have been a stale request and we ignore it.
        if ((entry != null) && (entry.requestID == reqId))
        {
          // ASSERT.isTrue(entry.cache != null)
  
          // The assertion basically means that we should only receive
          // NOTFLUSHED if the original query was AREYOUFLUSHED.  The
          // only legal reply to REQUESTFLUSH is, eventually, FLUSHED.
  
          // Our original query was for "are you flushed" and we now
          // have the reply, so:
          // 1) remove the entry from the flush map and disable the alarm
          // 2) handle the new prefixes and replay any cached data messages.
          flushMap.remove(streamID);
          entry.resend.cancel();
  
          StreamSet streamSet = getStreamSetForMessage(cMsg);
  
          if(streamSet == null)
          {
            //check that we don't already have a streamSet for this source, because there
            // should only be the one at any one time (except for links, see below...)
            SIBUuid8 sourceMEUuid = cMsg.getGuaranteedSourceMessagingEngineUUID();
            if(sourceMap.containsValue(sourceMEUuid))
            {
              //if we do, try to remove it
              Iterator<SIBUuid12> itr = sourceMap.keySet().iterator();
              while(itr.hasNext())
              {
                SIBUuid12 sid = itr.next();
                SIBUuid8 uuid = sourceMap.get(sid);
                if(sourceMEUuid.equals(uuid))
                {
                  // 538529
                  // If this is a link it is possible to have multiple streams of messages
                  // coming in from the source ME concurrently. We have one for all point-to-point
                  // messages and one per topicspace that is sending messages to us.
                  // Ideally we need to check with the source before we flush any of our existing
                  // streams, as they may still be valid. We would do this by asking the sender if
                  // they are flushed (commented out code below). Unfortunately we can't do that because
                  // the sender (the other end of the link) may not be the source of the messages
                  // so it's not guaranteed to actually know about this stream even if it still
                  // exists (for example the ME's just been restarted and lost the non-persistent
                  // state for this stream). So, instead we simply keep all streams.
                  // TODO This isn't ideal so maybe we can fix it in the future 
                  if(destination.isLink())
                  {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                      SibTr.debug(tc, "WARNING: Existing Link stream, " + sid + ", not flushed");
                    
  //                  StreamSet existingStreamSet = (StreamSet)streamSets.get(sid);
  //
  //                  // Create and store the new request record
  //                  long reqID = messageProcessor.nextTick();
  //                  FlushQueryRecord newEntry = new FlushQueryRecord(sourceMEUuid,
  //                                               existingStreamSet.getDestUuid(),
  //                                               existingStreamSet.getBusUuid(),
  //                                               null,
  //                                               reqID);
  //                  flushMap.put(sid, newEntry);
  //
  //                  // Create and send the query
  //                  upControl.sendAreYouFlushedMessage(sourceMEUuid,
  //                                                     existingStreamSet.getDestUuid(),
  //                                                     existingStreamSet.getBusUuid(),
  //                                                     reqID,
  //                                                     sid);
  //
  //                  // Start the alarm.  The context for the alarm is the stream id, which
  //                  // we can use to look up the FlushQueryRecord if the alarm expires.
  //                  newEntry.resend = am.create(GDConfig.FLUSH_QUERY_INTERVAL, this, sid);
                  }
                  // If we're not a link we only ever expect a single streamSet to be in use
                  // by the source at any one time, so flush any that we already have.
                  else
                  {
                    flush(sid);
                    //there should only ever be at most one other streamSet found
                    //so stop looking if we found one.
                    break;
                  }
                }
              }
            }
            
            // If this is a link, get the actual target destination 
            String linkTarget = SIMPConstants.PTOP_TARGET_STREAM;
            JsDestinationAddress routingDest = cMsg.getRoutingDestination();
            if (destination.isLink() &&  routingDest!= null)
            {
              // get the destination and check if it is a topicspace
              DestinationHandler dest = 
                destination.getDestinationManager().
                  getDestinationInternal(routingDest.getDestinationName(),
                                         routingDest.getBusName(), true);
              if (dest!=null && dest.isPubSub())
                linkTarget = routingDest.getDestinationName();
            }           
            
            //create a new streamSet
            SIBUuid8 busId = cMsg.getGuaranteedCrossBusSourceBusUUID();
            if( busId == null)
            {
              busId = messageProcessor.getMessagingEngineBusUuid();
            }
            streamSet = addNewStreamSet(streamID,
                                        sourceMEUuid,
                                        cMsg.getGuaranteedTargetDestinationDefinitionUUID(),
                                        busId,
                                        linkTarget);
            //set all the completed prefixes
            long[] completedPrefixes = cMsg.getCompletedPrefixTicks();
            int[] priority  = cMsg.getCompletedPrefixPriority();
            int[] reliability = cMsg.getCompletedPrefixQOS();
            for(int i=0;i<completedPrefixes.length;i++)
            {
              streamSet.setPersistentData(priority[i],
                                                  Reliability.getReliability(reliability[i]),
                                                  completedPrefixes[i]);
            }
          }
          // Now replay any cached messages
          for(int i=0; i<entry.cacheIndex; i++)
          {
            handleMessage(entry.cache[i]);
          }
        }
      }
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "handleNotFlushedMessage");
    }