docs: private final ConsumerSession createSession(
                                                  final SIDestinationAddress destination)
                      throws SIDurableSubscriptionAlreadyExistsException,
                      SIDurableSubscriptionNotFoundException,
                      SIDurableSubscriptionMismatchException,
                      SIConnectionDroppedException, SIConnectionUnavailableException,
                      SIConnectionLostException, SILimitExceededException,
                      SINotAuthorizedException, SIDestinationLockedException,
                      SITemporaryDestinationNotFoundException, SIResourceException,
                      SIErrorException, SIIncorrectCallException,
                      SINotPossibleInCurrentConfigurationException {
  
          final String methodName = "createSession";
          if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled()) {
              SibTr.entry(this, TRACE, methodName, new Object[] { destination });
          }
  
          /*
           * Determine session properties
           */
  
          final SibRaEndpointConfiguration endpointConfiguration = _connection
                          .getEndpointConfiguration();
  
          final Reliability reliability = null;
          _unrecoverableReliability = _connection
                          .getEndpointActivation().isEndpointMethodTransactional() ? Reliability.NONE
                          : Reliability.BEST_EFFORT_NONPERSISTENT;
  
          final boolean enableReadAhead;
          final boolean noLocal = false;
  
          final DestinationType destinationType = endpointConfiguration
                          .getDestinationType();
  
          final String selector = endpointConfiguration.getMessageSelector();
          final String discriminator = endpointConfiguration.getDiscriminator();
          final SelectionCriteria selectionCriteria = _selectionCriteriaFactory
                          .createSelectionCriteria(discriminator, selector,
                                                   endpointConfiguration.getSelectorDomain());
  
          // Distributed will also set this to true in order that max active messages can be used
          // by processor. Max active messages is only adhered to when bifurcatable is set to true 
          // (It was initially only used by z/os), for now we'll always set this to true so that
          // processor will adhere to max active messages. We'll leave the variable here as opposed
          // to passing the value true directly in to the create consumer methods in case processor
          // change to use max active messages for non bifurcatable sessions.
          final boolean bifurcatable = true;
          final String alternateUser = null;
  
          /*
           * Create consumer session
           */
  
          final SICoreConnection connection = _connection.getConnection();
  
          ConsumerSession session = null;
  
          if (endpointConfiguration.isDurableSubscription()) {
  
              /*
               * Durable subscription
               */
  
              final String subscriptionName = endpointConfiguration
                              .getDurableSubscriptionName();
  
              final String durableSubscriptionHome = endpointConfiguration
                              .getDurableSubscriptionHome();
  
              final String subscriptionDurability = endpointConfiguration.getSubscriptionDurability();
  
              //default value non sharable
              boolean supportsMultipleConsumers=false;
  
              //So just check whether subscriptionDurability is shared durable subscription or not.
              //if yes set supportsMultipleConsumers to true. otherwise leave as it is i.e false
              if (subscriptionDurability.equalsIgnoreCase("DurableShared"))
                  supportsMultipleConsumers = true;
  
              /*
               * By default, read ahead is only supported for unshared
               * subscriptions
               */
  
              if (SibRaReadAhead.DEFAULT.equals(endpointConfiguration
                              .getReadAhead())) {
                  enableReadAhead = !supportsMultipleConsumers;
              } else {
                  enableReadAhead = SibRaReadAhead.ON
                                  .equals(endpointConfiguration.getReadAhead());
              }
  
              try {
  
                  session = connection
                                  .createConsumerSessionForDurableSubscription(
                                                                               subscriptionName, durableSubscriptionHome,
                                                                               destination, selectionCriteria,
                                                                               supportsMultipleConsumers, noLocal,
                                                                               reliability, enableReadAhead,
                                                                               _unrecoverableReliability, bifurcatable,
                                                                               alternateUser);
  
              } catch (final SIDurableSubscriptionNotFoundException exception) {
  
                  // No FFDC code needed
  
                  if (TraceComponent.isAnyTracingEnabled() && TRACE.isEventEnabled()) {
                      SibTr.exception(this, TRACE, exception);
                  }
  
                  /*
                   * Durable subscription doesn't exist so create it now and
                   * attempt to create consumer again
                   */
  
                  connection.createDurableSubscription(subscriptionName,
                                                       durableSubscriptionHome, destination,
                                                       selectionCriteria, supportsMultipleConsumers, noLocal,
                                                       alternateUser);
  
                  session = connection
                                  .createConsumerSessionForDurableSubscription(
                                                                               subscriptionName, durableSubscriptionHome,
                                                                               destination, selectionCriteria,
                                                                               supportsMultipleConsumers, noLocal,
                                                                               reliability, enableReadAhead,
                                                                               _unrecoverableReliability, bifurcatable,
                                                                               alternateUser);
  
              } catch (final SIDurableSubscriptionMismatchException exception) {
  
                  // No FFDC code needed
  
                  if (TraceComponent.isAnyTracingEnabled() && TRACE.isEventEnabled()) {
                      SibTr.exception(this, TRACE, exception);
                  }
  
                  /*
                   * Durable subscription exists but doesn't match the given
                   * parameters so delete it, create a new one and attempt to
                   * create consumer again
                   */
  
                  connection.deleteDurableSubscription(subscriptionName,
                                                       durableSubscriptionHome);
  
                  connection.createDurableSubscription(subscriptionName,
                                                       durableSubscriptionHome, destination,
                                                       selectionCriteria, supportsMultipleConsumers, noLocal,
                                                       alternateUser);
  
                  session = connection
                                  .createConsumerSessionForDurableSubscription(
                                                                               subscriptionName, durableSubscriptionHome,
                                                                               destination, selectionCriteria,
                                                                               supportsMultipleConsumers, noLocal,
                                                                               reliability, enableReadAhead,
                                                                               _unrecoverableReliability, bifurcatable,
                                                                               alternateUser);
  
              }
  
          } else {
  
              /*
               * Queue, port, service or non-durable topic
               */
  
              //subscriptionName is needed for shared non-durable subscriptions
              final String subscriptionName = endpointConfiguration.getDurableSubscriptionName();
  
              //get subscriptionDurability from config. Possibilities are NonDurable/NonDurableShared
              final String subscriptionDurability = endpointConfiguration.getSubscriptionDurability();
  
              boolean isNonDurableShared = false;
              if (subscriptionDurability.equalsIgnoreCase("NonDurableShared"))
                  isNonDurableShared = true;
  
              /*
               * By default, enable read ahead only for non-durable non-shared topics
               */
  
              if (SibRaReadAhead.DEFAULT.equals(endpointConfiguration
                              .getReadAhead())) {
                  enableReadAhead = (DestinationType.TOPICSPACE
                                  .equals(destinationType)) && !isNonDurableShared;
              } else {
                  enableReadAhead = SibRaReadAhead.ON
                                  .equals(endpointConfiguration.getReadAhead());
              }
  
              if (!isNonDurableShared) {
  
                  // AllowMessageGathering is a config option on JMSQueues, if we have a value
                  // then try to use it, if not use the method call that does not support
                  // this field. When connecting to an older ME then we will get a SIIncorrectCallException
                  // at which point we should revert to the older style method call.
                  String allowMessageGathering = _connection.getEndpointConfiguration().getAllowMessageGathering();
                  //Either Non-Durable Non Shared Topic or Queue consumer. Leaving the old code as it is 
                  if (allowMessageGathering != null)
                  {
                      boolean allow = (allowMessageGathering.equals(ApiJmsConstants.GATHER_MESSAGES_ON));
                      try
                      {
                          session = connection.createConsumerSession(destination,
                                                                     destinationType, selectionCriteria, reliability,
                                                                     enableReadAhead, noLocal, _unrecoverableReliability,
                                                                     bifurcatable, alternateUser, true, allow, null);
                      } catch (SIIncorrectCallException ex)
                      {
                          // No FFDC code needed
                          // This will happen when we connect to an old ME, in this case rever
                          // to the old method call.
                          session = connection.createConsumerSession(destination,
                                                                     destinationType, selectionCriteria, reliability,
                                                                     enableReadAhead, noLocal, _unrecoverableReliability,
                                                                     bifurcatable, alternateUser);
                      }
                  }
                  else
                  {
                      session = connection.createConsumerSession(destination,
                                                                 destinationType, selectionCriteria, reliability,
                                                                 enableReadAhead, noLocal, _unrecoverableReliability,
                                                                 bifurcatable, alternateUser);
                  }
              } else {
                  //Non Durable Shared consumer. Use the new API createSharedConsumerSession
                  session = connection.createSharedConsumerSession(subscriptionName,
                                                                   destination,
                                                                   destinationType,
                                                                   selectionCriteria,
                                                                   reliability,
                                                                   enableReadAhead,
                                                                   true, //supportMultipleConsumers
                                                                   noLocal,
                                                                   _unrecoverableReliability,
                                                                   bifurcatable,
                                                                   alternateUser,
                                                                   true, //ignoreInitialIndoubts is true by default
                                                                   false, //allowMessageGathering is false for topic
                                                                   null // message control properties, passing null. Not used.
                  );
  
              }
  
          }
  
          if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled()) {
              SibTr.exit(this, TRACE, methodName, session);
          }
          return session;
  
      }