@Test public void computeOptimizedScanOrAndCombinedFilter() {
    KafkaScanTrimmer kafkaScanTrimmer = new KafkaScanTrimmer(fullHouse, null);
    // partition = 0 and 30 <= offset < 35 or partition = 3 and 35 <= offset < 75  or (partition = 0 and offset = 40)

    ExprNodeGenericFuncDesc
        part1 =
        and(Lists.newArrayList(greaterThanEq(Lists.newArrayList(offsetColumn, thirtyLong)),
            eq(Lists.newArrayList(partitionColumn, zeroInt)),
            lessThan(Lists.newArrayList(offsetColumn, thirtyFiveLong))));

    ExprNodeGenericFuncDesc
        part2 =
        and(Lists.newArrayList(greaterThanEq(Lists.newArrayList(offsetColumn, thirtyFiveLong)),
            eq(Lists.newArrayList(partitionColumn, threeInt)),
            lessThan(Lists.newArrayList(offsetColumn, seventyFiveLong))));

    ExprNodeGenericFuncDesc
        part3 =
        and(Lists.newArrayList(eq(Lists.newArrayList(offsetColumn, fortyLong)),
            eq(Lists.newArrayList(partitionColumn, zeroInt))));

    ExprNodeGenericFuncDesc orExpression = or(Lists.newArrayList(part1, part2, part3));

    assertNotNull(orExpression);
    Map
        actual =
        kafkaScanTrimmer.computeOptimizedScan(SerializationUtilities
            .deserializeExpression(SerializationUtilities.serializeExpression(orExpression)));
    TopicPartition tpZero = new TopicPartition(topic, 0);
    TopicPartition toThree = new TopicPartition(topic, 3);
    KafkaInputSplit split1 = new KafkaInputSplit(topic, 0, 30, 41, PATH);
    KafkaInputSplit split2 = new KafkaInputSplit(topic, 3, 35, 75, PATH);

    Map expected = ImmutableMap.of(tpZero, split1, toThree, split2);
    Assert.assertEquals(expected, actual);

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