Commit the specified offsets for the last consumed message.

@param msg
@param groupId
@return {@code true} if the topic is in subscription list, {@code false}
otherwise
@since 1.3.2
333786
public boolean commmitAsync(KafkaMessage msg, String groupId) {
        KafkaMsgConsumer kafkaConsumer = cacheConsumers.get(groupId);
        return kafkaConsumer != null ? kafkaConsumer.commitAsync(msg) : false;
    }
906171
private void doGroupEnd(final JsonObject message) {
    String groupID = message.getString("group");
    DefaultConnectionInputGroup group = groups.remove(groupID);
    if (group != null) {
      Object args = deserializer.deserialize(message);
      if (log.isDebugEnabled()) {
        log.debug(String.format("%s - Group ended: Group[group=%s, args=%s]", this, group.id(), args));
      }
      group.handleEnd(args);
    }
  }
305120
public long mpNews(int group, String mpNews) {
		return send(new Filter(false, String.valueOf(group)), null, "mpnews", mpNews);
	}
255912
public void commitAddActiveMessage()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this,tc, "commitAddActiveMessage");
  
    // If we're a member of a ConsumerSet, inform them of the commit
    if(classifyingMessages)
      consumerSet.commitAddActiveMessage();    

    if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(this,tc, "commitAddActiveMessage");
  }
554359
private static boolean isDataMessage(final ByteBuf msg) {
    return DcpMutationMessage.is(msg) || DcpDeletionMessage.is(msg) || DcpExpirationMessage.is(msg);
  }
1237587
public int send(final LogMsgGroup group) throws IOException {
		Preconditions.checkNotNull(group);

		executeMask(group);
		executeSkipJsonTag(group);

		HttpClient httpClient = new HttpClient(apiConfig);

		// retransmit any logs on the resend queue

		resendQueue.drain(httpClient, LOG_SAVE_PATH, true);

		// convert to json bytes

		byte[] jsonBytes = objectMapper.writer().writeValueAsBytes(group);

		// post to stackify

		int statusCode = HttpURLConnection.HTTP_INTERNAL_ERROR;

		try {
			httpClient.post(LOG_SAVE_PATH, jsonBytes, true);
			statusCode = HttpURLConnection.HTTP_OK;
		} catch (IOException t) {
			LOGGER.info("Queueing logs for retransmission due to IOException");
			resendQueue.offer(jsonBytes, t);
			throw t;
		} catch (HttpException e) {
			statusCode = e.getStatusCode();
			LOGGER.info("Queueing logs for retransmission due to HttpException", e);
			resendQueue.offer(jsonBytes, e);
		}

		return statusCode;
	}
1138961
private boolean takeMessage(SIMPMessage msg, boolean isOnItemStream)
                    throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "takeMessage", new Object[] { this, msg, Boolean.valueOf(isOnItemStream) });

        boolean msgTaken = false;

        // Put should only have been called if we said we were Ready but
        // we'll double check we are still Ready as there is a window where
        // we could have stopped/closed/ended the receive after the CD
        // had picked us from the ready list.
        if ((_ready && (_keyGroup == null)) || (_keyGroup != null) && (_keyGroup.isGroupReady()))
        {
            //Consumer is still ready so it can take the message
            //indicate that we're not Ready for any more messages
            //setReadyState(false);
            _ready = false;

            if (_keyGroup != null)
                _keyGroup.groupNotReady();

            // Before we can take the message we need to know that this consumer or its
            // ConsumerSet will allow it (they haven't reached their maxActiveMessage limits).
            // If they will accept it we have to, in the case of a ConsumerSet, 'reserve' the
            // space upfront because multiple members of the set could be concurrently adding
            // active messages. If this add happens to take the last available slot in the set's
            // quota it will keep the set write-locked for the duration of the message lookup. During
            // this time no other consumers in the set will be able to add any more messages.
            // This is done to ensure that the set's suspend/resume state is consistent and we don't
            // have to try to un-suspend consumers when we find there isn't a message to take after all.
            boolean msgAccepted = prepareAddActiveMessage();
            if (!msgAccepted)
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(tc, "The consumer set is already suspended ");
            }
            else
            {
                boolean activeMessageAttached = false;
                // Whatever the outcome we must commit or rollback the prepareAddActiveMessage()
                try
                {
                    //we HAVE taken this message
                    msgTaken = true;

                    //attach the message we have taken
                    try
                    {
                        activeMessageAttached = attachAndLockMsg(msg, isOnItemStream);
                    } catch (SIResourceException e)
                    {
                        // No FFDC code needed
                        SibTr.exception(tc, e);

                        activeMessageAttached = false;
                        checkWaiting();

                        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                            SibTr.exit(tc, "takeMessage", e);

                        throw e;
                    }

                    // We found an attached message, therefore we have another active
                    // message so commit the prepare we did above.
                    if (activeMessageAttached)
                        commitAddActiveMessage();

                    checkWaiting();
                } finally
                {
                    // If a message wasn't attached the commit won't have happened, instead
                    // roll back the prepare of the add, as if the prepare never happened.
                    if (!activeMessageAttached)
                        rollbackAddActiveMessage();
                }
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "takeMessage", Boolean.valueOf(msgTaken));

        return msgTaken;
    }
1061238
def encode_offset_commit_request_kafka(cls, group, payloads):
        
        return kafka.protocol.commit.OffsetCommitRequest[2](
            consumer_group=group,
            consumer_group_generation_id=kafka.protocol.commit.OffsetCommitRequest[2].DEFAULT_GENERATION_ID,
            consumer_id='',
            retention_time=kafka.protocol.commit.OffsetCommitRequest[2].DEFAULT_RETENTION_TIME,
            topics=[(
                topic,
                [(
                    partition,
                    payload.offset,
                    payload.metadata)
                    for partition, payload in six.iteritems(topic_payloads)])
                for topic, topic_payloads in six.iteritems(group_by_topic_and_partition(payloads))])
968123
public boolean removeMessageListener(String consumerGroupId, String topic,
            IKafkaMessageListener messageListener) {
        KafkaMsgConsumer kafkaConsumer = cacheConsumers.get(consumerGroupId);
        return kafkaConsumer != null ? kafkaConsumer.removeMessageListener(topic, messageListener)
                : false;
    }
1367414
public final void commit( boolean commitGets , List<String> deliveredMessageIDs ) throws JMSException
    {
    	if (!transacted)
            throw new IllegalStateException("Session is not transacted"); // [JMS SPEC]
    	
    	externalAccessLock.readLock().lock();
    	try
		{
	    	checkNotClosed();   
	        commitUpdates(commitGets,deliveredMessageIDs,true);
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
13553
protected synchronized void groupCloseNotification(Conversation conversation, 
                                                      ProxyQueueConversationGroup group)
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(tc, "groupCloseNotification", new Object[] {conversation, group});
      if (convToGroupMap.remove(conversation) == null)
      {
         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(tc, "group unknown!");
      }
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "groupCloseNotification");
   }
1003351
public final void satisfiedRequest(long tick, SIMPMessage msg)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "satisfiedRequest", new Object[] { Long.valueOf(tick), msg });

    boolean sendMsg = false;
    // used to send a message outside the synchronized block
    long startstamp = tick;
    AOValue value = null;
    boolean transitionToFlushed = false;

    synchronized (this)
    {
      try
      {
        if (startedFlush || closed)
        { // unlock the message
          msg.unlockMsg(msg.getLockID(),null,true);
          // change this tick to final
          TickRange r = new TickRange(TickRange.Completed, tick, tick);
          stream.writeCompleted(r);
          transitionToFlushed = tryTransitionToFlushed();
        }
        else
        {

          stream.setCursor(tick);
          TickRange r = stream.getNext();
          if (r.type != TickRange.Requested)
          {
            // this tick is no longer requested
            // unlock the message
            msg.unlockMsg(msg.getLockID(),null,true);
          }
          else
          {
            AORequested requested = (AORequested) r.value;
            long waitTime = System.currentTimeMillis() - requested.startTime;
            if (msg
              .getReliability()
              .compareTo(Reliability.RELIABLE_NONPERSISTENT)
              <= 0)
            {
              // Note that this means that the message storage strategy in the MS is either
              // STORE_NEVER or STORE_MAYBE. In either case, we create an AOValue tick with
              // storage strategy equal to STORE_NEVER. In fact, we don't even add this AOValue
              // tick to an ItemStream. Also, we do not persistently lock the message

              int reliability = msg.getReliability().getIndex();
              int priority = msg.getMessage().getPriority().intValue();
              // message is already locked. change tick to value state

              value =
                new AOValue(
                  tick,
                  msg,
                  msg.getID(),
                  AbstractItem.STORE_NEVER,
                  0L,
                  waitTime,
                  latestTick[reliability][priority]);
              latestTick[reliability][priority] = tick; // update latestTick

              TickRange r2 = TickRange.newValueTick(tick, value, 0L);

              stream.writeCombinedRange(r2);
              r2 = stream.findCompletedRange(r2);
              // now start the DecisionExpected timer
              dem.addTimeoutEntry(value);
              // send message if active
              if (active)
              {
                sendMsg = true;
                startstamp = r2.startstamp;
              }
            }
            else
            {
              int storagePolicy;
              if (msg
                .getReliability()
                .compareTo(Reliability.ASSURED_PERSISTENT)
                < 0)
                storagePolicy = AbstractItem.STORE_EVENTUALLY;
              else
                storagePolicy = AbstractItem.STORE_ALWAYS;

              // change the AORequested tick to INSERTING state
               ((AORequested) r.value).inserting = true;

              // message needs to be persistently locked, and tick persisted
              int reliability = msg.getReliability().getIndex();
              int priority = msg.getMessage().getPriority().intValue();
              PersistLockAndTick update =
                new PersistLockAndTick(
                  tick,
                  msg,
                  storagePolicy,
                  waitTime,
                  latestTick[reliability][priority]);
              latestTick[reliability][priority] = tick; // update latestTick

              parent.getPersistLockThread().enqueueWork(update);
              countAsyncUpdatesOutstanding++;
            }
          }
        }
      }
      catch (MessageStoreException e)
      {
        // MessageStoreException shouldn't occur so FFDC.
        FFDCFilter.processException(
          e,
          "com.ibm.ws.sib.processor.impl.AOStream.satisfiedRequest",
          "1:1058:1.80.3.24",
          this);

        SibTr.exception(tc, e);
      }
      catch (Exception e)
      {
        // probably a bug - log error
        FFDCFilter.processException(
          e,
          "com.ibm.ws.sib.processor.impl.AOStream.satisfiedRequest",
          "1:1069:1.80.3.24",
          this);
        SIErrorException e2 =
          new SIErrorException(
            nls.getFormattedMessage(
              "INTERNAL_MESSAGING_ERROR_CWSIP0002",
              new Object[] {
                "com.ibm.ws.sib.processor.impl.AOStream",
                "1:1077:1.80.3.24",
                e },
              null));
        SibTr.exception(tc, e2);
        SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
          new Object[] {
            "com.ibm.ws.sib.processor.impl.AOStream",
            "1:1084:1.80.3.24",
            e });
      }
    } // end synchronized (this)

    if (sendMsg)
    {
      long prevTick = value.getPrevTick();
      parent.sendRemoteGetData(
        msg,
        remoteMEUuid,
        gatheringTargetDestUuid,
        streamId,
        prevTick,
        startstamp,
        tick,
        value.getWaitTime());

      msg.releaseJsMessage();
    }
    if (transitionToFlushed)
    {
      parent.sendFlushed(remoteMEUuid, gatheringTargetDestUuid, streamId);
    }
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "satisfiedRequest");

  }
85666
@Override
    public void registerForEvents(SIMPMessage msg)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "registerForEvents", msg);

        msg.registerMessageEventListener(MessageEvents.EXPIRY_NOTIFICATION, this);
        msg.registerMessageEventListener(
                                         MessageEvents.REFERENCES_DROPPED_TO_ZERO,
                                         this);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "registerForEvents");
    }
377829
void doGroupEnd(String group, Object args) {
    checkOpen();
    JsonObject message = createMessage(args)
        .putString("action", "endGroup")
        .putString("group", group);
    if (open && !paused) {
      if (log.isDebugEnabled()) {
        log.debug(String.format("%s - Group end: Group[group=%s, args=%s]", this, group, args));
      }
      eventBus.send(inAddress, message);
    }
    groups.remove(group);
  }
573316
public boolean removeMessageListener(String topic, IKafkaMessageListener msgListener) {
        synchronized (topicMsgListeners) {
            if (topicMsgListeners.remove(topic, msgListener)) {
                if (topicMsgListeners.get(topic).isEmpty()) {
                    // no more listener, stop worker
                    KafkaMsgConsumerWorker worker = topicWorkers.remove(topic);
                    if (worker != null) {
                        worker.stopWorker();
                    }
                }
                return true;
            }
        }
        return false;
    }
1315819
@Override
    public boolean requeueSilent(IQueueMessage<ID, DATA> _msg) throws QueueException.QueueIsFull {
        IQueueMessage<ID, DATA> msg = _msg.clone();
        putToQueue(msg);
        if (!isEphemeralDisabled()) {
            ephemeralStorage.remove(msg.getId());
        }
        return true;
    }
735195
private synchronized boolean msgCanBeSent( long stamp, boolean nackMsg )
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "msgCanBeSent", new Object[] { Long.valueOf(stamp),
          Boolean.valueOf(nackMsg),
          Long.valueOf(firstMsgOutsideWindow),
          Boolean.valueOf(containsGuesses)});

    boolean sendMessage = true;
    
    // Check whether we can send the message
    // Don't send any messages once there is a guess in the stream (unless a NACK)
    if ((stamp >= firstMsgOutsideWindow) || (containsGuesses && !nackMsg))
    {
      sendMessage = false;
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
      {
        SibTr.debug(tc, "firstMsgOutsideWindow is: " + firstMsgOutsideWindow
        + " sendWindow is " +  sendWindow + " containsGuesses is " + containsGuesses);
      }
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "msgCanBeSent", Boolean.valueOf(sendMessage));

    return sendMessage;
  }