com.networknt.tram.test

Class KafkaTestBase

java.lang.Object

com.networknt.tram.test.KafkaTestBase

public abstract class KafkaTestBase
extends Object

The base for the Kafka tests. It brings up:

• A ZooKeeper mini cluster
• Three Kafka Brokers (mini clusters)

Code in this test is based on the following GitHub repository: https://github.com/sakserv/hadoop-mini-clusters (ASL licensed), as per commit bc6b2b2d5f6424d5f377aa6c0871e82a956462ef

Field Summary

Fields

Modifier and Type	Field and Description
protected static String	brokerConnectionStrings
protected static KafkaTestEnvironment	kafkaServer
protected static org.slf4j.Logger	LOG
protected static int	NUMBER_OF_KAFKA_SERVERS
protected static Properties	secureProps
protected static Properties	standardProps
static org.junit.rules.TemporaryFolder	tempFolder
protected static scala.concurrent.duration.FiniteDuration	timeout

Constructor Summary

Constructors

Constructor and Description
KafkaTestBase()

Method Summary

Modifier and Type	Method and Description
protected void	assertAtLeastOnceForTopic(Properties properties, String topic, int partition, Set<Integer> expectedElements, long timeoutMillis) We manually handle the timeout instead of using JUnit's timeout to return failure instead of timeout error.
protected void	assertExactlyOnceForTopic(Properties properties, String topic, int partition, List<Integer> expectedElements)
protected void	assertExactlyOnceForTopic(Properties properties, String topic, int partition, List<Integer> expectedElements, long timeoutMillis) We manually handle the timeout instead of using JUnit's timeout to return failure instead of timeout error.
protected static void	createTestTopic(String topic, int numberOfPartitions, short replicationFactor)
protected static void	deleteTestTopic(String topic)
static void	prepare()
static void	prepare(boolean hideKafkaBehindProxy)
protected static void	shutdownClusters()
static void	shutDownServices()
protected static void	startClusters(boolean secureMode, boolean hideKafkaBehindProxy)

Methods inherited from class java.lang.Object

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

Field Detail

LOG

protected static final org.slf4j.Logger LOG

NUMBER_OF_KAFKA_SERVERS

protected static final int NUMBER_OF_KAFKA_SERVERS

See Also:

Constant Field Values

brokerConnectionStrings

protected static String brokerConnectionStrings

standardProps

protected static Properties standardProps

timeout

protected static scala.concurrent.duration.FiniteDuration timeout

kafkaServer

protected static KafkaTestEnvironment kafkaServer

tempFolder

public static org.junit.rules.TemporaryFolder tempFolder

secureProps

protected static Properties secureProps

Constructor Detail

KafkaTestBase

public KafkaTestBase()

Method Detail

prepare

public static void prepare()
                    throws ClassNotFoundException

Throws:

ClassNotFoundException

prepare

public static void prepare(boolean hideKafkaBehindProxy)
                    throws ClassNotFoundException

Throws:

ClassNotFoundException

shutDownServices

public static void shutDownServices()
                             throws Exception

Throws:

Exception

startClusters

protected static void startClusters(boolean secureMode,
                                    boolean hideKafkaBehindProxy)
                             throws ClassNotFoundException

Throws:

ClassNotFoundException

shutdownClusters

protected static void shutdownClusters()
                                throws Exception

Throws:

Exception

createTestTopic

protected static void createTestTopic(String topic,
                                      int numberOfPartitions,
                                      short replicationFactor)

deleteTestTopic

protected static void deleteTestTopic(String topic)

assertAtLeastOnceForTopic

protected void assertAtLeastOnceForTopic(Properties properties,
                                         String topic,
                                         int partition,
                                         Set<Integer> expectedElements,
                                         long timeoutMillis)

We manually handle the timeout instead of using JUnit's timeout to return failure instead of timeout error. After timeout we assume that there are missing records and there is a bug, not that the test has run out of time.

Parameters:

properties - Properties

topic - String topic

partition - int partition

expectedElements - Set

timeoutMillis - long timeout

assertExactlyOnceForTopic

protected void assertExactlyOnceForTopic(Properties properties,
                                         String topic,
                                         int partition,
                                         List<Integer> expectedElements)

assertExactlyOnceForTopic

protected void assertExactlyOnceForTopic(Properties properties,
                                         String topic,
                                         int partition,
                                         List<Integer> expectedElements,
                                         long timeoutMillis)

We manually handle the timeout instead of using JUnit's timeout to return failure instead of timeout error. After timeout we assume that there are missing records and there is a bug, not that the test has run out of time.

Parameters:

properties - Properties

topic - String topic

partition - int partition

expectedElements - List

timeoutMillis - long timeout