org.apache.flume.shared.kafka.test

Class KafkaPartitionTestUtil

java.lang.Object

org.apache.flume.shared.kafka.test.KafkaPartitionTestUtil

public class KafkaPartitionTestUtil
extends Object

Field Summary

Fields

Modifier and Type	Field and Description
static String	PARTITION_HEADER

Constructor Summary

Constructors

Constructor and Description
KafkaPartitionTestUtil()

Method Summary

Modifier and Type	Method and Description
static void	checkResultsAgainstSkew(PartitionTestScenario scenario, Map<Integer,List<Event>> partitionMap, Map<Integer,List<byte[]>> resultsMap, int staticPtn, int numMsgs) This method checks the retrieved messages (passed as resultsMap) against the expected results (passed as partitionMap).
static List<Event>	generateSkewedMessageList(PartitionTestScenario scenario, int numMsgs, Map<Integer,List<Event>> partitionMap, int numPtns, int staticPtn) This method is can be used to create a list of events for use in Kafka partition tests.
static Map<Integer,List<byte[]>>	retrieveRecordsFromPartitions(String topic, int numPtns, Properties consumerProperties) Return a map containing one List of records per partition.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

PARTITION_HEADER

public static final String PARTITION_HEADER

See Also:

Constant Field Values

Constructor Detail

KafkaPartitionTestUtil

public KafkaPartitionTestUtil()

Method Detail

checkResultsAgainstSkew

public static void checkResultsAgainstSkew(PartitionTestScenario scenario,
                                           Map<Integer,List<Event>> partitionMap,
                                           Map<Integer,List<byte[]>> resultsMap,
                                           int staticPtn,
                                           int numMsgs)

This method checks the retrieved messages (passed as resultsMap) against the expected results (passed as partitionMap). The behaviour of this method is slightly different depending on the scenario: - STATIC_HEADER_ONLY: Don't check partitionMap, just check that all messages are in the passed staticPtn partition. - NO_PARTITION_HEADERS: Check that messages are evenly distributed between partitions (requires numMsgs to be a multiple of the number of partitons) - else: Check that the contents of each partition list in resultsMap is the same as that specified in partitionMap. As this is a testing method, it will issue JUnit AssertionExceptions if the results are not as expected.

Parameters:

scenario -

partitionMap -

resultsMap -

staticPtn -

numMsgs -

generateSkewedMessageList

public static List<Event> generateSkewedMessageList(PartitionTestScenario scenario,
                                                    int numMsgs,
                                                    Map<Integer,List<Event>> partitionMap,
                                                    int numPtns,
                                                    int staticPtn)

This method is can be used to create a list of events for use in Kafka partition tests. Depending on the scenario it will deliberate generate an artificially skewed distribution of events-per-partition (populated into the passed partitionMap) and then ordered randomly into the resulting List of events. Four scenarios are catered for: - STATIC_HEADER_ONLY: All events are put into the partition specified by the staticPtn param - PARTITION_ID_HEADER_ONLY: Events are skewed into three partitions - STATIC_HEADER_AND_PARTITION_ID: Events are skewed into two partitions, with all others going into the partition specified by staticPtn - NO_PARTITION_HEADERS: No partition header is set and the partitionMap is not populated

Parameters:

scenario - The scenario being catered for.

numMsgs - The number of messages to generate

partitionMap - A map of Integer (partitionId) and List of Events - to be populated

numPtns - The number of partitions to be populated.

staticPtn - The static partition to be assigned to.

Returns:

retrieveRecordsFromPartitions

public static Map<Integer,List<byte[]>> retrieveRecordsFromPartitions(String topic,
                                                                      int numPtns,
                                                                      Properties consumerProperties)

Return a map containing one List of records per partition. This internally creates a Kafka Consumer using the provided consumer properties.

Parameters:

numPtns -

consumerProperties -

Returns:

A Map of Partitions(Integer) and the resulting List of messages (byte[]) retrieved