@Nullable private static Map<TopicPartition, KafkaInputSplit> buildScanForTimesPredicate(
      Map<TopicPartition, KafkaInputSplit> fullHouse,
      PredicateLeaf.Operator operator,
      long timestamp,
      boolean flip,
      boolean negation,
      KafkaConsumer<byte[], byte[]> consumer) {
    long
        increment =
        (flip && operator == PredicateLeaf.Operator.LESS_THAN
            || negation && operator == PredicateLeaf.Operator.LESS_THAN_EQUALS) ? 1L : 0L;
    // only accepted cases are timestamp_column [ > ; >= ; = ]constant
    if (operator == PredicateLeaf.Operator.EQUALS || flip ^ negation) {
      final Map<TopicPartition, Long> timePartitionsMap = Maps.toMap(fullHouse.keySet(), tp -> timestamp + increment);
      try {
        // Based on Kafka docs
        // NULL will be returned for that partition If the message format version in a partition is before 0.10.0
        Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp = consumer.offsetsForTimes(timePartitionsMap);
        return Maps.toMap(fullHouse.keySet(), tp -> {
          KafkaInputSplit existing = fullHouse.get(tp);
          OffsetAndTimestamp foundOffsetAndTime = offsetAndTimestamp.get(tp);
          //Null in case filter doesn't match or field not existing ie old broker thus return empty scan.
          final long startOffset = foundOffsetAndTime == null ? existing.getEndOffset() : foundOffsetAndTime.offset();
          return new KafkaInputSplit(Objects.requireNonNull(tp).topic(),
              tp.partition(),
              startOffset,
              existing.getEndOffset(),
              existing.getPath());
        });
      } catch (Exception e) {
        LOG.error("Error while looking up offsets for time", e);
        //Bailout when can not figure out offsets for times.
        return null;
      }

    }
    return null;
  }
@param fullScan    full kafka scan to be pruned
@param operator    operator kind
@param offsetConst offset constant value
@param flip        true if position of constant and column were flipped by default assuming COLUMN OP CONSTANT
@param negation    true if the expression is a negation of the original expression

@return optimized kafka scan
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaScanTrimmer.java