code: public void commit(KafkaTopicPartitionOffset... tpoList) {
          Map<String, Map<TopicPartition, OffsetAndMetadata>> topicOffsets = new HashMap<>();
          for (KafkaTopicPartitionOffset tpo : tpoList) {
              Map<TopicPartition, OffsetAndMetadata> offsets = topicOffsets.get(tpo.topic);
              if (offsets == null) {
                  offsets = new HashMap<>();
                  topicOffsets.put(tpo.topic, offsets);
              }
              TopicPartition tp = new TopicPartition(tpo.topic, tpo.partition);
              OffsetAndMetadata oam = new OffsetAndMetadata(tpo.offset);
              offsets.put(tp, oam);
          }
          for (Entry<String, Map<TopicPartition, OffsetAndMetadata>> entry : topicOffsets
                  .entrySet()) {
              String topic = entry.getKey();
              KafkaConsumer<String, byte[]> consumer = _getConsumer(topic);
              synchronized (consumer) {
                  Set<String> subscription = consumer.subscription();
                  if (subscription == null || !subscription.contains(topic)) {
                      // this consumer has not subscribed to the topic
                      LOGGER.warn("Not subscribed to topic [" + topic + "] yet!");
                  } else {
                      try {
                          consumer.commitSync(entry.getValue());
                      } catch (WakeupException e) {
                      } catch (Exception e) {
                          throw new KafkaException(e);
                      }
                  }
              }
          }
      }
docs: Commit the specified offsets for the specified list of topics and
  partitions.
  
  @param tpoList
  @since 1.3.2