@Test public void computeOptimizedScanPartitionBinaryOpFilter() {
    KafkaScanTrimmer kafkaScanTrimmer = new KafkaScanTrimmer(fullHouse, null);
    int partitionId = 2;
    ExprNodeDesc constant = ConstantExprBuilder.build(partitionId);
    final List<ExprNodeDesc> children = Lists.newArrayList(partitionColumn, constant);

    ExprNodeGenericFuncDesc node = eq(children);
    assertNotNull(node);

    Map
        actual =
        kafkaScanTrimmer.computeOptimizedScan(SerializationUtilities
            .deserializeExpression(SerializationUtilities.serializeExpression(node)));
    Map expected = Maps.filterValues(fullHouse, tp -> Objects.requireNonNull(tp).getPartition() == partitionId);
    Assert.assertEquals(expected, actual);

    ExprNodeGenericFuncDesc lessNode = lessThan(children);
    assertNotNull(lessNode);
    actual =
        kafkaScanTrimmer.computeOptimizedScan(SerializationUtilities
            .deserializeExpression(SerializationUtilities.serializeExpression(lessNode)));
    expected = Maps.filterValues(fullHouse, tp -> Objects.requireNonNull(tp).getPartition() < partitionId);
    Assert.assertEquals(expected, actual);

    ExprNodeGenericFuncDesc lessEqNode = lessThanEq(children);

    assertNotNull(lessEqNode);
    actual =
        kafkaScanTrimmer.computeOptimizedScan(SerializationUtilities
            .deserializeExpression(SerializationUtilities.serializeExpression(lessEqNode)));
    expected = Maps.filterValues(fullHouse, tp -> Objects.requireNonNull(tp).getPartition() <= partitionId);
    Assert.assertEquals(expected, actual);

  }
Test Class for Kafka Trimmer Class.
kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaScanTrimmerTest.java