The second flow is: from JMS 2.0 API createSharedConsumer - for 'non durable shared'
544279
private MessageConsumer createConsumer(Destination destination, String name, String messageSelector, boolean noLocal,
                                           boolean isSharedNonDurable) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createConsumer", new Object[] { destination, messageSelector, noLocal, isSharedNonDurable });
        JmsMsgConsumer consumer = null;

        checkNotClosed();
        if (destination == null) {
            throw (InvalidDestinationException) JmsErrorUtils.newThrowable(InvalidDestinationException.class,
                                                                           "INVALID_VALUE_CWSIA0048",
                                                                           new Object[] { "Destination", "null" },
                                                                           tc
                            );
        }

        if (isSharedNonDurable) {
            if ((name == null) || ("".equals(name))) {
                throw (JMSException) JmsErrorUtils.newThrowable(JMSException.class,
                                                                "INVALID_VALUE_CWSIA0048",
                                                                new Object[] { "sharedSubscriptionName", name },
                                                                tc
                                );
            }
        }

        // Check that this is not a foreign implementation.
        // Note that in some situations this method call can return a _different_
        // destination than the one passed in as the parameter (e.g. Spring).
        JmsDestinationImpl nativeDest = JmsDestinationImpl.checkNativeInstance(destination);

        // Check that this destination is not blocked from creating consumers
        JmsDestinationImpl.checkBlockedStatus(nativeDest);

        // create the consumer properties object filling in as much information
        // as we have available at the moment.
        ConsumerProperties newProps = new ConsumerProperties(nativeDest,
                        null, // DestType                filled in later
                        messageSelector, // Selector
                        null, // Reliability             filled in later
                        false, // ReadAhead               filled in later
                        true, // RecovExpress            filled in later
                        noLocal, // NoLocal
                        name, // SubName  .. with JMS 2.0, the value is set for all cases
                        connection.getClientID(), // ClientID .. with JMS 2.0, the value is set for all cases
                        isSharedNonDurable, // SupportMultiple .. with JMS 2.0, the value is set for all cases         
                        null // DurableSubscriptionHome n/a for non-durable consumers
        );

        // Hold the state lock while the msg consumer constructor is called to prevent
        // it from reading invalid start/stopped state while start is in progress. This
        // will basically cause and createConsumer calls on this session to wait until
        // session.start is complete.
        synchronized (stateLock) {
            consumer = instantiateConsumer(newProps);
            // newly created consumers must be synchronous
            syncConsumers.add(consumer);

            // d353701 - output a warning message if there are 'lots' of producers active.
            int totalConsumers = syncConsumers.size() + asyncConsumers.size();
            if (totalConsumers % PROD_CONS_WARNING_THRESHOLD == 0) {
                // We wish to tell the user which line of their application created the session,
                // so we must obtain a line of stack trace from their application.
                String errorLocation = JmsErrorUtils.getFirstApplicationStackString();
                SibTr.warning(tc, "MANY_CONSUMERS_WARNING_CWSIA0059", new Object[] { "" + totalConsumers, errorLocation });
            }

        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "createConsumer", consumer);
        return consumer;
    }
651538
private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer origMessage, boolean retain, Integer messageID) {
        Log.trace("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
        if (Log.DEBUG) {
            Log.trace("content <{}>", DebugUtils.payload2Str(origMessage));
            Log.trace("subscription tree {}", subscriptions.dumpTree());
        }
        for (final Subscription sub : subscriptions.matches(topic)) {
            if(m_clientIDs.get(sub.getClientId()) == null) {
                subscriptions.removeSubscription(topic, sub.getClientId());
                continue;
            }
            if (qos.ordinal() > sub.getRequestedQos().ordinal()) {
                qos = sub.getRequestedQos();
            }

            ByteBuffer message = origMessage.duplicate();

            if (qos == AbstractMessage.QOSType.MOST_ONE && sub.isActive()) {
                //QoS 0
                sendPublish(sub.getClientId(), topic, qos, message, false);
            } else {
                //QoS 1 or 2
                //if the target subscription is not clean session and is not connected => store it
                if (!sub.isCleanSession() && !sub.isActive()) {
                    //clone the event with matching clientID
                    PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID);
                    m_storageService.storePublishForFuture(newPublishEvt);
                } else  {
                    //if QoS 2 then store it in temp memory
                    if (qos ==AbstractMessage.QOSType.EXACTLY_ONCE) {
                        String publishKey = String.format("%s%d", sub.getClientId(), messageID);
                        PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID);
                        m_storageService.addInFlight(newPublishEvt, publishKey);
                    }
                    //publish
                    if (sub.isActive()) {
                        sendPublish(sub.getClientId(), topic, qos, message, false);
                    }
                }
            }
        }
    }
759243
@Beta
    public static <T, U> BiConsumer<T, U> sc(final Object mutex, final BiConsumer<T, U> biConsumer) {
        N.checkArgNotNull(mutex, "mutex");
        N.checkArgNotNull(biConsumer, "biConsumer");

        return new BiConsumer<T, U>() {
            @Override
            public void accept(T t, U u) {
                synchronized (mutex) {
                    biConsumer.accept(t, u);
                }
            }
        };
    }
598911
@Override
    public java.util.concurrent.Future<DescribeReservedNodeOfferingsResult> describeReservedNodeOfferingsAsync(
            com.amazonaws.handlers.AsyncHandler<DescribeReservedNodeOfferingsRequest, DescribeReservedNodeOfferingsResult> asyncHandler) {

        return describeReservedNodeOfferingsAsync(new DescribeReservedNodeOfferingsRequest(), asyncHandler);
    }
1152728
@Override
    public String getShareDurableSubscriptions() {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "getShareDurableSubscriptions");
        String val = jcaConnectionFactory.getShareDurableSubscriptions();
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "getShareDurableSubscriptions", val);
        return val;
    }
1368834
private ServiceRegistration<?> registerMBeanService(String jmsResourceName, BundleContext bundleContext) {
        Dictionary<String, String> props = new Hashtable<String, String>();
        props.put(KEY_SERVICE_VENDOR, "IBM");

        JmsServiceProviderMBeanImpl jmsProviderMBean = new JmsServiceProviderMBeanImpl(getServerName(), KEY_JMS2_PROVIDER);
        props.put(KEY_JMX_OBJECTNAME, jmsProviderMBean.getobjectName());

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "JmsQueueMBeanImpl=" + jmsProviderMBean.getobjectName() + " props=" + props);
        }
        return bundleContext.registerService(JmsServiceProviderMBeanImpl.class, jmsProviderMBean, props);
    }
976822
private void setAsyncListener(MessageListener listener) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "setAsyncListener", listener);

        consumer = new Consumer(listener, session, sessionAckMode);
        try {
            // ConsumerSession must be stopped to register the callback
            coreConsumerSession.stop();
            int batchSize = 1;
            boolean deliverImmediately = false;
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(this, tc, "Registering async callback for consumer" + consumer);
            // the register call will replace any existing callback, so no need for
            // separate deregister call.
            coreConsumerSession.registerAsynchConsumerCallback(consumer,
                                                               0, // maxActiveMessages - no limit
                                                               0, // messageLockExpiry - no expiry
                                                               batchSize,
                                                               session.getOrderingContext()
                            );

            // Deal with the case where the listener is set after the connection
            // is started. Note that this will handle the restarting of the core
            // consumer session if we paused it earlier in this block.
            if (session.getState() == STARTED)
                coreConsumerSession.start(deliverImmediately);
        } catch (SIIncorrectCallException sie) {
            // No FFDC code needed
            // d222942 review. Default message ok.
            // d238447 FFDC Review. FFDC here, see comment in next catch for more detail.
            throw (JMSException) JmsErrorUtils.newThrowable(JMSException.class,
                                                            "EXCEPTION_RECEIVED_CWSIA0085",
                                                            new Object[] { sie, "JmsMsgConsumerImpl.setAsyncListener" },
                                                            sie,
                                                            "JmsMsgConsumerImpl.setAsyncListener#1",
                                                            this,
                                                            tc);
        } catch (SIException sie) {
            // No FFDC code needed
            // The exceptions thrown by stop, registerAsynchConsumerCallback and start are mostly
            // the "connection lost" family. registerAsynchConsumerCallback can also throw an
            // SIIncorrectCallException if the coreConsumerSession hasn't been stopped, but that
            // shouldn't be possible in the code above. Default message ok.
            // d238447 FFDC Review. SIIncorrectCall separated into its own catch block above,
            //   no FFDCs for this group.
            throw (JMSException) JmsErrorUtils.newThrowable(JMSException.class,
                                                            "EXCEPTION_RECEIVED_CWSIA0085",
                                                            new Object[] { sie, "JmsMsgConsumerImpl.setAsyncListener" },
                                                            sie,
                                                            null, // null probeId = no FFDC
                                                            this,
                                                            tc);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "setAsyncListener");
    }
407425
protected DispatchableKey createConsumerKey(DispatchableConsumerPoint consumerPoint,
                                            SelectionCriteria criteria,
                                            SIBUuid12 connectionUuid,
                                            boolean readAhead,
                                            boolean forwardScanning,
                                            JSConsumerSet consumerSet) throws SISelectorSyntaxException, SIDiscriminatorSyntaxException, SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "createConsumerKey", new Object[] { consumerPoint,
                                                          criteria,
                                                          connectionUuid,
                                                          new Boolean(readAhead),
                                                          new Boolean(forwardScanning),
                                                          consumerSet});

    RemoteQPConsumerKey consKey =
      new RemoteQPConsumerKey(consumerPoint,
                            this,criteria,
                            connectionUuid,
                            readAhead,
                            forwardScanning);
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "createConsumerKey", consKey);
    return consKey;
  }
800880
@Override
	public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException
    {
        return createDurableSubscriber(topic,name,null,false);
    }
543353
protected void buildShareWar() throws MojoExecutionException {
        buildCustomWarInDir(SHARE_WAR_PREFIX_NAME, shareModules,
                alfrescoGroupId, alfrescoShareWarArtifactId, alfrescoShareVersion);

        copyShareLog4jConfig();
        copyShareConfigCustom();
        copyHotswapAgentProperties(SHARE_WAR_PREFIX_NAME);


        String shareWarArtifactId = packageAndInstallCustomWar(SHARE_WAR_PREFIX_NAME);

        // Set up the custom share war to be run by Tomcat plugin
        runnerAlfrescoGroupId = "${project.groupId}";
        runnerAlfrescoShareWarArtifactId = shareWarArtifactId;
        runnerAlfrescoShareVersion = "${project.version}";
    }
971405
private void createProxyHandler() throws MessageStoreException {
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.entry(tc, "createProxyHandler");

        _multiMEProxyHandler = new MultiMEProxyHandler(this, _txManager);
        _persistentStore.addItemStream(_multiMEProxyHandler, _txManager
                        .createAutoCommitTransaction());

        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.exit(tc, "createProxyHandler");
    }
1342757
@Override
    public ConsumerManager createConsumerManager()
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "createConsumerManager");

        ConsumerDispatcher consumerDispatcher = (ConsumerDispatcher) getOutputHandler();
        if (consumerDispatcher == null)
        {
            consumerDispatcher = new ConsumerDispatcher(destinationHandler, this, new ConsumerDispatcherState());
            setOutputHandler(consumerDispatcher);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createConsumerManager", consumerDispatcher);
        return consumerDispatcher;
    }
1199686
private SimpleConsumerThread<T> createAndStartSimpleConsumerThread(
			List<KafkaTopicPartitionState<TopicAndPartition>> seedPartitions,
			Node leader,
			ExceptionProxy errorHandler) throws IOException, ClassNotFoundException {
		// each thread needs its own copy of the deserializer, because the deserializer is
		// not necessarily thread safe
		final KafkaDeserializationSchema<T> clonedDeserializer =
				InstantiationUtil.clone(deserializer, runtimeContext.getUserCodeClassLoader());

		// seed thread with list of fetch partitions (otherwise it would shut down immediately again
		SimpleConsumerThread<T> brokerThread = new SimpleConsumerThread<>(
				this, errorHandler, kafkaConfig, leader, seedPartitions, unassignedPartitionsQueue,
				clonedDeserializer, invalidOffsetBehavior);

		brokerThread.setName(String.format("SimpleConsumer - %s - broker-%s (%s:%d)",
				runtimeContext.getTaskName(), leader.id(), leader.host(), leader.port()));
		brokerThread.setDaemon(true);
		brokerThread.start();

		LOG.info("Starting thread {}", brokerThread.getName());
		return brokerThread;
	}
587465
@Override
	public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	        checkNotClosed();
	        
	        RemoteTopicSubscriber subscriber =  new RemoteTopicSubscriber(idProvider.createID(),
	        		                                                      this,
	                                                                      DestinationTools.asRef(topic),
	                                                                      messageSelector,
	                                                                      noLocal);
	        registerConsumer(subscriber);
	        subscriber.remoteInit();
	        return subscriber;
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
754811
@Override
    public ConsumerSession createConsumerSessionForDurableSubscription(String subscriptionName,
                                                                       boolean enableReadAhead,
                                                                       Reliability unrecoverableReliability,
                                                                       boolean bifurcatable)

                    throws SIConnectionUnavailableException, SIConnectionDroppedException,
                    SIErrorException,
                    SIDurableSubscriptionNotFoundException, SIDurableSubscriptionMismatchException,
                    SIDestinationLockedException, SIIncorrectCallException, SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConnection.tc.isEntryEnabled())
            SibTr.entry(
                        CoreSPIConnection.tc,
                        "createConsumerSessionForDurableSubscription",
                        new Object[] {
                                      this,
                                      subscriptionName,
                                      Boolean.valueOf(enableReadAhead),
                                      unrecoverableReliability,
                                      Boolean.valueOf(bifurcatable) });

        // Do not need to check that the destination information is correct.
        // As we are using the subscription data straight from the named subscription

        // Find the ConsumerDispatcher for this subscription
        // Note that it must be homed on this ME
        HashMap durableSubs = _destinationManager.getDurableSubscriptionsTable();

        ConsumerDispatcher cd = null;
        synchronized (durableSubs)
        {
            //Look up the consumer dispatcher for this subId in the system durable subs list
            cd =
                            (ConsumerDispatcher) durableSubs.get(subscriptionName);

            // Check that the durable subscription exists
            if (cd == null)
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription");

                throw new SIDurableSubscriptionNotFoundException(
                                nls.getFormattedMessage(
                                                        "SUBSCRIPTION_DOESNT_EXIST_ERROR_CWSIP0072",
                                                        new Object[] { subscriptionName,
                                                                      _messageProcessor.getMessagingEngineName() },
                                                        null));
            }
        }

        //Get the destinationHandler from the ConsumerDispatcher
        //Note that we know it is PubSub  
        DestinationHandler destination = cd.getDestination();
        SIDestinationAddress destinationAddress = SIMPUtils.createJsDestinationAddress(
                                                                                       destination.getName(),
                                                                                       _messageProcessor.getMessagingEngineUuid());

        //Get the state object representing the subscription from the ConsumerDispatcher
        ConsumerDispatcherState subState = cd.getConsumerDispatcherState();

        ConsumerSessionImpl consumer = null;

        // Synchronize on the close object, we don't want the connection closing
        // while we try to add the subscription.
        synchronized (this)
        {
            // See if this connection has been closed
            checkNotClosed();

            //create the consumer session for this subscription
            try
            {
                consumer =
                                new ConsumerSessionImpl(destination,
                                                destinationAddress,
                                                subState,
                                                this,
                                                enableReadAhead,
                                                false,
                                                unrecoverableReliability,
                                                bifurcatable,
                                                true,
                                                false);
            } catch (SINotPossibleInCurrentConfigurationException e)
            {
                // No FFDC code needed
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription", "SIErrorException");
                throw new SIErrorException(e);
            } catch (SISessionUnavailableException e)
            {
                // FFDC
                FFDCFilter
                                .processException(
                                                  e,
                                                  "com.ibm.ws.sib.processor.impl.ConnectionImpl.createConsumerSessionForDurableSubscription",
                                                  "1:2129:1.347.1.25",
                                                  this);

                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                          "1:1809:1.285",
                                          e });

                // This should never be thrown
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription", "SIErrorException");
                throw new SIErrorException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                                                      "1:2146:1.347.1.25",
                                                                      e },
                                                        null),
                                e);
            } catch (SITemporaryDestinationNotFoundException e)
            {
                // FFDC
                FFDCFilter
                                .processException(
                                                  e,
                                                  "com.ibm.ws.sib.processor.impl.ConnectionImpl.createConsumerSessionForDurableSubscription",
                                                  "1:2159:1.347.1.25",
                                                  this);

                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                          "1:2165:1.347.1.25",
                                          e });

                // This should never be thrown
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription", "SIErrorException");
                throw new SIErrorException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                                                      "1:2176:1.347.1.25",
                                                                      e },
                                                        null),
                                e);
            } catch (SINonDurableSubscriptionMismatchException e)
            {
                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                          "1:1953:1.347.1.25",
                                          e });

                // This should never be thrown
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription", "SIErrorException");
                throw new SIErrorException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                                                      "1:1964:1.347.1.25",
                                                                      e },
                                                        null),
                                e);
            }

            synchronized (_consumers)
            {
                //store a reference
                _consumers.add(consumer);
            }

            _messageProcessor.addConsumer(consumer);
        }

        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConnection.tc.isEntryEnabled())
            SibTr.exit(CoreSPIConnection.tc, "createConsumerSessionForDurableSubscription", consumer);
        return consumer;
    }
967704
private Service createServiceHook(String flowletName, Iterable<ConsumerSupplier<?>> consumerSuppliers,
                                    AtomicReference<FlowletProgramController> controller) {
    final List<String> streams = Lists.newArrayList();
    for (ConsumerSupplier<?> consumerSupplier : consumerSuppliers) {
      QueueName queueName = consumerSupplier.getQueueName();
      if (queueName.isStream()) {
        streams.add(queueName.getSimpleName());
      }
    }

    // If no stream, returns a no-op Service
    if (streams.isEmpty()) {
      return new AbstractService() {
        @Override
        protected void doStart() {
          notifyStarted();
        }

        @Override
        protected void doStop() {
          notifyStopped();
        }
      };
    }
    return new FlowletServiceHook(flowletName, streams, controller);
  }
576978
@Override
	public void unregisterConsumer(LocalMessageConsumer consumer)
    {
        super.unregisterConsumer(consumer);
        if (!consumer.isDurable())
        {
            log.debug("Removing non-durable subscription "+consumer.getSubscriberId());
            subscriptionsLock.writeLock().lock();
            try
    		{
            	LocalTopicSubscription subscription = subscriptionMap.remove(consumer.getSubscriberId());
            	if (subscription != null)
            		removeSubscription(subscription);
    		}
            finally
            {
            	subscriptionsLock.writeLock().unlock();
            }
        }
    }