@Override
  public void deleteMessages(Map<SystemStreamPartition, String> offsets) {
    if (deleteCommittedMessages) {
      Map<TopicPartition, RecordsToDelete> recordsToDelete = offsets.entrySet()
          .stream()
          .collect(Collectors.toMap(entry ->
            new TopicPartition(entry.getKey().getStream(), entry.getKey().getPartition().getPartitionId()),
            entry -> RecordsToDelete.beforeOffset(Long.parseLong(entry.getValue()) + 1)));

      adminClient.deleteRecords(recordsToDelete).all().whenComplete((ignored, exception) -> {
        if (exception != null) {
          LOG.error("Delete message failed for SSPs " + offsets.keySet() + " due to", exception);
        }
      });

      deleteMessageCalled = true;
    }
  }
Delete records up to (and including) the provided ssp offsets for
all system stream partitions specified in the map.
This only works with Kafka cluster 0.11 or later. Otherwise it's a no-op.
@param offsets specifies up to what offsets the messages should be deleted
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java