

package org.apache.flink.streaming.connectors.kafka;

import org.apache.flink.test.util.SecureTestEnvironment;

import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Kafka Secure Connection (kerberos) IT test case.
 */
public class Kafka09SecuredRunITCase extends KafkaConsumerTestBase {

	protected static final Logger LOG = LoggerFactory.getLogger(Kafka09SecuredRunITCase.class);

	@BeforeClass
	public static void prepare() throws ClassNotFoundException {
		LOG.info("-------------------------------------------------------------------------");
		LOG.info("    Starting Kafka09SecuredRunITCase ");
		LOG.info("-------------------------------------------------------------------------");

		SecureTestEnvironment.prepare(tempFolder);
		SecureTestEnvironment.populateFlinkSecureConfigurations(getFlinkConfiguration());

		startClusters(true);
	}

	@AfterClass
	public static void shutDownServices() {
		shutdownClusters();
		SecureTestEnvironment.cleanup();
	}

	//timeout interval is large since in Travis, ZK connection timeout occurs frequently
	//The timeout for the test case is 2 times timeout of ZK connection
	@Test(timeout = 600000)
	public void testMultipleTopics() throws Exception {
		runProduceConsumeMultipleTopics();
	}

}
