Merge pull request #41 from Aiven-Open/update-testkit

added overrides to kafak startup

Author: aindriu-aiven

kafka-testkit/src/main/java/io/aiven/commons/kafka/testkit/KafkaConnectRunner.java

@@ -148,10 +148,10 @@ public String getGroupId() {
 	 * @throws IOException
 	 *             if listener ports can not be found.
 	 */
-	public void startConnectCluster(final String clusterName, final Class<? extends Connector> connectorClass)
-			throws IOException {
+	public void startConnectCluster(final String clusterName, final Class<? extends Connector> connectorClass,
+			Map<String, String> configOverrides) throws IOException {
 		final List<Integer> ports = findListenerPorts();
-		startConnectCluster(clusterName, ports.get(0), ports.get(1), connectorClass);
+		startConnectCluster(clusterName, ports.get(0), ports.get(1), connectorClass, configOverrides);
 	}
 
 	/**
@@ -167,7 +167,7 @@ public void startConnectCluster(final String clusterName, final Class<? extends
 	 *            the class for the connector.
 	 */
 	public void startConnectCluster(final String clusterName, final int localPort, final int containerPort,
-			final Class<? extends Connector> connectorClass) {
+			final Class<? extends Connector> connectorClass, Map<String, String> configOverrides) {
 		this.clusterName = clusterName;
 		this.containerListenerPort = containerPort;
 		final Properties brokerProperties = new Properties();
@@ -178,7 +178,7 @@ public void startConnectCluster(final String clusterName, final int localPort, f
 		brokerProperties.put("listener.security.protocol.map", "PLAINTEXT:PLAINTEXT,TESTCONTAINERS:PLAINTEXT");
 
 		connectCluster = new EmbeddedConnectCluster.Builder().name(clusterName).brokerProps(brokerProperties)
-				.workerProps(getWorkerProperties(connectorClass)).numWorkers(1).build();
+				.workerProps(getWorkerProperties(connectorClass, configOverrides)).numWorkers(1).build();
 		connectCluster.start();
 		LOGGER.info("connectCluster {} started", clusterName);
 	}
@@ -286,14 +286,15 @@ public void restartConnector(final String connectorName) {
 	 *            the connector class to start, may be {@code null}.
 	 * @return the default set of worker properties.
 	 */
-	public Map<String, String> getWorkerProperties(final Class<? extends Connector> connectorClass) {
+	public Map<String, String> getWorkerProperties(final Class<? extends Connector> connectorClass,
+			final Map<String, String> configOverrides) {
 		Map<String, String> workerProperties = new HashMap<>();
 		workerProperties.put(ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG, ByteArrayConverter.class.getName());
 		workerProperties.put(ConnectorConfig.VALUE_CONVERTER_CLASS_CONFIG, ByteArrayConverter.class.getCanonicalName());
 		workerProperties.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG,
 				Long.toString(offsetFlushInterval.toMillis()));
 		workerProperties.put("plugin.discovery", "HYBRID_WARN");
-
+		workerProperties.putAll(configOverrides);
 		if (connectorClass != null) {
 			workerProperties.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, connectorClass.getName());
 		}

kafka-testkit/src/main/java/io/aiven/commons/kafka/testkit/KafkaIntegrationTestBase.java

@@ -30,6 +30,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Locale;
+import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Supplier;
@@ -181,9 +182,9 @@ void captureTestInfo(final TestInfo testInfo) {
 	 * @throws InterruptedException
 	 *             on interrupted thread.
 	 */
-	final protected KafkaManager setupKafka(final Class<? extends Connector> connectorClass)
-			throws IOException, ExecutionException, InterruptedException {
-		return setupKafka(false, connectorClass);
+	final protected KafkaManager setupKafka(final Class<? extends Connector> connectorClass,
+			Map<String, String> configOverrides) throws IOException, ExecutionException, InterruptedException {
+		return setupKafka(false, connectorClass, configOverrides);
 	}
 
 	/**
@@ -201,8 +202,8 @@ final protected KafkaManager setupKafka(final Class<? extends Connector> connect
 	 * @throws IOException
 	 *             on IO error.
 	 */
-	final protected KafkaManager setupKafka(final boolean forceRestart, final Class<? extends Connector> connectorClass)
-			throws IOException {
+	final protected KafkaManager setupKafka(final boolean forceRestart, final Class<? extends Connector> connectorClass,
+			Map<String, String> configOverrides) throws IOException {
 		KafkaManager kafkaManager = KAFKA_MANAGER_THREAD_LOCAL.get();
 		if (kafkaManager != null && forceRestart) {
 			tearDownKafka();
@@ -212,7 +213,7 @@ final protected KafkaManager setupKafka(final boolean forceRestart, final Class<
 			final String clusterName = new CasedString(CasedString.StringCase.CAMEL,
 					testInfo.getTestClass().get().getSimpleName()).toCase(CasedString.StringCase.KEBAB)
 					.toLowerCase(Locale.ROOT);
-			kafkaManager = new KafkaManager(clusterName, getOffsetFlushInterval(), connectorClass);
+			kafkaManager = new KafkaManager(clusterName, getOffsetFlushInterval(), connectorClass, configOverrides);
 			KAFKA_MANAGER_THREAD_LOCAL.set(kafkaManager);
 		}
 		return kafkaManager;
@@ -301,11 +302,12 @@ protected final <K> void waitForStorage(final Duration timeout, final Supplier<C
 	 * @param <K>
 	 *            the data type of the storage value. (must implement equals).
 	 */
+	@SuppressWarnings("PMD.ClassCastExceptionWithToArray")
 	protected final <K> void waitForStorage(final Duration timeout, final Supplier<Collection<K>> storageList,
 			final Collection<K> expectedStorage) {
 		// wait for them to show up.
 		await().atMost(timeout).pollInterval(Duration.ofSeconds(1)).untilAsserted(() -> {
-			assertThat(storageList.get()).containsExactlyInAnyOrderElementsOf(expectedStorage);
+			assertThat(storageList.get()).contains((K[]) expectedStorage.toArray());
 		});
 	}
 }

kafka-testkit/src/main/java/io/aiven/commons/kafka/testkit/KafkaManager.java

@@ -61,9 +61,9 @@ public final class KafkaManager {
 	 *             if the cluster can not be started.
 	 */
 	public KafkaManager(final String clusterName, final Duration offsetFlushInterval,
-			final Class<? extends Connector> connectorClass) throws IOException {
+			final Class<? extends Connector> connectorClass, Map<String, String> configOverrides) throws IOException {
 		connectRunner = new KafkaConnectRunner(offsetFlushInterval);
-		connectRunner.startConnectCluster(clusterName, connectorClass);
+		connectRunner.startConnectCluster(clusterName, connectorClass, configOverrides);
 
 		final Map<String, Object> adminClientConfig = new HashMap<>();
 		adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, connectRunner.getBootstrapServers());