Let's use an integration test to experiment with topic setup

Author: solsson

build.gradle

@@ -1,3 +1,4 @@
+
 repositories {
   mavenCentral()
   jcenter()
@@ -14,10 +15,15 @@ apply plugin: "jacoco"
 
 group 'se.yolean'
 
+sourceCompatibility = 1.8
+
 apply plugin: 'application'
 mainClassName = 'se.yolean.kafka.topic.client.cli.Client'
 
-sourceCompatibility = 1.8
+configurations {
+  compile.exclude group: 'org.slf4j', module: 'slf4j-log4j12'
+  compile.exclude group: 'log4j', module: 'log4j'
+}
 
 dependencies {
   compile group: 'javax.inject', name: 'javax.inject', version: '1'
@@ -31,6 +37,7 @@ dependencies {
   compile group: 'io.confluent', name: 'kafka-schema-registry-client', version: '4.0.0'
 
   runtime group: 'org.slf4j', name: 'slf4j-simple', version: '1.7.25'
+  runtime group: 'org.slf4j', name: 'log4j-over-slf4j', version: '1.7.25'
   compile group: 'structlog4j', name: 'structlog4j-api', version: '1.0.0'
   compile group: 'structlog4j', name: 'structlog4j-json', version: '1.0.0'
 
@@ -55,6 +62,12 @@ buildscript {
   }
   dependencies {
     classpath 'org.standardout:gradle-eclipseconfig:1.1.0'
+    classpath 'org.unbroken-dome.gradle-plugins:gradle-testsets-plugin:1.4.2'
   }
 }
 apply plugin: 'org.standardout.eclipseconfig'
+
+apply plugin: 'org.unbroken-dome.test-sets'
+testSets {
+  itest
+}

src/itest/java/se/yolean/kafka/topic/client/service/IntegrationTestConfigLocalhost.java

@@ -0,0 +1,23 @@
+package se.yolean.kafka.topic.client.service;
+
+import java.util.Properties;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.name.Names;
+
+public class IntegrationTestConfigLocalhost extends AbstractModule {
+
+  @Override
+  protected void configure() {
+    bind(String.class).annotatedWith(Names.named("config:bootstrap")).toInstance("localhost:9092");
+
+    bind(String.class).annotatedWith(Names.named("config:adminTopic")).toInstance("_topic_declarations");
+
+    bind(Integer.class).annotatedWith(Names.named("config:adminInitTimeoutMs")).toInstance(1000);
+
+    bind(Integer.class).annotatedWith(Names.named("config:adminTopicDesiredReplicationFactor")).toInstance(1);
+
+    bind(Properties.class).annotatedWith(Names.named("admin")).toProvider(AdminClientPropsProvider.class);
+  }
+
+}

src/itest/java/se/yolean/kafka/topic/client/service/TopicDeclarationsTopicCheckTest.java

@@ -33,15 +33,9 @@ public void tearDown() throws Exception {
 
   @Test
   public void test() throws Exception {
-    Injector injector = Guice.createInjector(new AbstractModule() {
-      @Override
-      protected void configure() {
-        bind(Properties.class).toInstance(new Properties());
-        //bind(TopicDeclarationsTopicCheck.class);
-      }
-    });
-    //TopicDeclarationsTopicCheck check = injector.getInstance(TopicDeclarationsTopicCheck.class);
-    //check.createOrVerifySchemaTopic();
+    Injector injector = Guice.createInjector(new IntegrationTestConfigLocalhost());
+    TopicDeclarationsTopicCheck check = injector.getInstance(TopicDeclarationsTopicCheck.class);
+    check.createOrVerifySchemaTopic();
   }
 
 }

src/itest/resources/simplelogger.properties

@@ -0,0 +1,2 @@
+org.slf4j.simpleLogger.defaultLogLevel=info
+org.slf4j.simpleLogger.log.se.yolean=debug

src/main/java/se/yolean/kafka/topic/client/service/AdminClientPropsProvider.java

@@ -6,30 +6,21 @@
 import javax.inject.Named;
 import javax.inject.Provider;
 
+import org.apache.kafka.clients.admin.AdminClientConfig;
+
 public class AdminClientPropsProvider implements Provider<Properties> {
 
 	private String bootstrap;
-	private String acks;
 
 	@Inject
-	public AdminClientPropsProvider(@Named("config:bootstrap") String bootstrap,
-			@Named("config:acks") String acks) {
+	public AdminClientPropsProvider(@Named("config:bootstrap") String bootstrap) {
 		this.bootstrap = bootstrap;
-		this.acks = acks;
 	}
 
 	@Override
 	public Properties get() {
-		// https://kafka.apache.org/0110/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html
 		Properties props = new Properties();
-		props.put("bootstrap.servers", bootstrap);
-		props.put("acks", acks);
-		props.put("retries", 0);
-		props.put("batch.size", 16384);
-		props.put("linger.ms", 1);
-		props.put("buffer.memory", 33554432);
-		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+		props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
 		return props;
 	}
 

src/test/resources/simplelogger.properties

@@ -0,0 +1,2 @@
+org.slf4j.simpleLogger.defaultLogLevel=info
+org.slf4j.simpleLogger.log.se.yolean=debug