code: public boolean commit(KafkaMessage msg) {
          KafkaConsumer<String, byte[]> consumer = _getConsumer(msg.topic());
          synchronized (consumer) {
              Set<String> subscription = consumer.subscription();
              if (subscription == null || !subscription.contains(msg.topic())) {
                  // this consumer has not subscribed to the topic
                  return false;
              } else {
                  try {
                      Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                      offsets.put(new TopicPartition(msg.topic(), msg.partition()),
                              new OffsetAndMetadata(msg.offset() + 1));
                      consumer.commitSync(offsets);
                  } catch (WakeupException e) {
                  } catch (Exception e) {
                      throw new KafkaException(e);
                  }
                  return true;
              }
          }
      }
docs: Commit the specified offsets for the last consumed message.
  
  @param msg
  @return {@code true} if the topic is in subscription list, {@code false}
  otherwise
  @since 1.3.2