Repository: kafka-ops/kafka-topology-builder
Pull request #169 (2020-12-14 10:57:38)

Add Kafka Streams integration test bed
Added integration test for Streams app ACLs. 

I also had to add a dependency for kafka-clients that I think was actually missing as first-order-deps in the first place. Had issues running without this when adding the kafka-streams dep.


Patch:
diff --git a/.github/workflows/nightly-artifacts-build.yml b/.github/workflows/nightly-artifacts-build.yml
new file mode 100644
index 0000000..cf3c917
--- /dev/null
+++ b/.github/workflows/nightly-artifacts-build.yml
@@ -0,0 +1,76 @@
+name: Nightly Builds
+on:
+  schedule:
+    - cron: '0 0 * * *'
+  push:
+    branches:
+      - master
+jobs:
+  ktb_docker_build:
+    name: Build docker image
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: docker/setup-buildx-action@v1
+      - name: Set up the JDK
+        uses: actions/setup-java@v1
+        with:
+          java-version: 1.8
+      - name: Cache dependencies
+        uses: actions/cache@v2
+        with:
+          path: ~/.m2
+          key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
+          restore-keys: ${{ runner.os }}-m2
+      - name: Build with Maven
+        run: mvn -B package --file pom.xml
+      - name: copy fat jar
+        run: cp ./target/kafka-topology-builder.jar release/docker
+      - name: copy runner file
+        run: cp ./src/main/scripts/kafka-topology-builder.sh release/docker
+      - name: list files (release dir)
+        run: ls -l
+        working-directory: release/docker
+      - name: Login to DockerHub
+        uses: docker/login-action@v1
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: Push to Docker Hub
+        uses: docker/build-push-action@v2
+        with:
+          context: release/docker
+          file: release/docker/Dockerfile
+          push: true
+          tags: purbon/kafka-topology-builder:latest
+  package_build:
+    name: Build rpm/deb packages (using maven)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up the JDK
+        uses: actions/setup-java@v1
+        with:
+          java-version: 1.8
+      - name: Cache dependencies
+        uses: actions/cache@v2
+        with:
+          path: ~/.m2
+          key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
+          restore-keys: ${{ runner.os }}-m2
+      - name: Build with Maven
+        run: mvn -B package --file pom.xml
+      - name: Build the rpm
+        run: mvn rpm:rpm
+      - uses: actions/upload-artifact@v2
+        with:
+          name: RPM package
+          path: target/rpm/kafka-topology-builder/RPMS/noarch/*.rpm
+      - uses: actions/upload-artifact@v2
+        with:
+          name: DEB package
+          path: target/*.deb
+      - uses: actions/upload-artifact@v2
+        with:
+          name: FAT jar
+          path: target/kafka-topology-builder.jar
diff --git a/.github/workflows/artifacts-build.yml b/.github/workflows/release-artifacts-build.yml
similarity index 91%
rename from .github/workflows/artifacts-build.yml
rename to .github/workflows/release-artifacts-build.yml
index 3d2b3dc..625235d 100644
--- a/.github/workflows/artifacts-build.yml
+++ b/.github/workflows/release-artifacts-build.yml
@@ -1,10 +1,8 @@
-name: Artifacts Builder
+name: Release Builds
 on:
-  schedule:
-    - cron: '0 0 * * *'
   push:
-    branches:
-      - master
+    tags:
+      - v*
 jobs:
   package_build:
     name: Build rpm/deb packages (using maven)
diff --git a/.github/workflows/docker.yml b/.github/workflows/release-docker.yml
similarity index 71%
rename from .github/workflows/docker.yml
rename to .github/workflows/release-docker.yml
index 5a0b13c..02c92f8 100644
--- a/.github/workflows/docker.yml
+++ b/.github/workflows/release-docker.yml
@@ -1,14 +1,24 @@
 name: Docker publish latest
 on:
-  schedule:
-    - cron: '0 0 * * *'
+  push:
+    tags:
+      - 'v*.*.*'
 jobs:
-  ktb_build:
+  ktb_docker_build:
     name: Build docker image
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
       - uses: docker/setup-buildx-action@v1
+      - name: Docker meta
+        id: docker_meta
+        uses: crazy-max/ghaction-docker-meta@v1
+        with:
+          images: purbon/kafka-topology-builder # list of Docker images to use as base name for tags
+          tag-sha: true
+          tag-semver: |
+              {{version}}
+              {{major}}.{{minor}}
       - name: Set up the JDK
         uses: actions/setup-java@v1
         with:
@@ -39,4 +49,5 @@ jobs:
           context: release/docker
           file: release/docker/Dockerfile
           push: true
-          tags: purbon/kafka-topology-builder:latest
+          tags: ${{ steps.docker_meta.outputs.tags }}
+          labels: ${{ steps.docker_meta.outputs.labels }}
diff --git a/pom.xml b/pom.xml
index e1bf35f..1702be4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -416,7 +416,7 @@
     <maven-failsafe-plugin.version>2.22.2</maven-failsafe-plugin.version>
     <!-- dependencies -->
     <jackson.version>2.10.3</jackson.version>
-    <kafka.version>2.4.1</kafka.version>
+    <kafka.version>2.5.0</kafka.version>
     <log4j.version>2.13.3</log4j.version>
     <httpclient.version>4.5.11</httpclient.version>
     <zookeeper.version>3.5.7</zookeeper.version>
@@ -481,6 +481,17 @@
       <artifactId>zookeeper</artifactId>
       <version>${zookeeper.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>${kafka.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-streams</artifactId>
+      <version>${kafka.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>io.confluent</groupId>
       <artifactId>kafka-client-plugins</artifactId>
@@ -567,5 +578,11 @@
       <version>${assertj.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <version>4.0.3</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>
diff --git a/src/deb/control/preinst b/src/deb/control/preinst
index b3304b4..3c98bd4 100644
--- a/src/deb/control/preinst
+++ b/src/deb/control/preinst
@@ -1,5 +1,11 @@
 #!/usr/bin/env bash
 
 mkdir -p /usr/local/kafka-topology-builder/bin
-addgroup --system ktb-kafka
-adduser --system --no-create-home  --ingroup ktb-kafka ktb-kafka
\ No newline at end of file
+
+if id "ktb-kafka" &>/dev/null;
+then
+  echo "User ktb-kafka already exist"
+else
+   addgroup --system ktb-kafka
+   adduser --system --no-create-home  --ingroup ktb-kafka ktb-kafka
+fi
diff --git a/src/deb/control/prerm b/src/deb/control/prerm
index ef5a02f..54f78ee 100644
--- a/src/deb/control/prerm
+++ b/src/deb/control/prerm
@@ -2,4 +2,9 @@
 
 rm -rf /usr/local/kafka-topology-builder/
 unlink /usr/bin/kafka-topology-builder
-userdel -f ktb-kafka
\ No newline at end of file
+userdel -f ktb-kafka
+
+if grep -q -E "^ktb-kafka:" /etc/group;
+then
+  groupdel ktb-kafka
+fi
\ No newline at end of file
diff --git a/src/main/scripts/preinstall.sh b/src/main/scripts/preinstall.sh
index 110f86a..e16f6a5 100644
--- a/src/main/scripts/preinstall.sh
+++ b/src/main/scripts/preinstall.sh
@@ -1,3 +1,8 @@
 #!/usr/bin/env bash
 
-adduser -M -r ktb-kafka
\ No newline at end of file
+if id "ktb-kafka" &>/dev/null;
+then
+  echo "User ktb-kafka already exist"
+else
+   adduser -M -r ktb-kafka
+fi
diff --git a/src/main/scripts/remove.sh b/src/main/scripts/remove.sh
index ef5a02f..54f78ee 100644
--- a/src/main/scripts/remove.sh
+++ b/src/main/scripts/remove.sh
@@ -2,4 +2,9 @@
 
 rm -rf /usr/local/kafka-topology-builder/
 unlink /usr/bin/kafka-topology-builder
-userdel -f ktb-kafka
\ No newline at end of file
+userdel -f ktb-kafka
+
+if grep -q -E "^ktb-kafka:" /etc/group;
+then
+  groupdel ktb-kafka
+fi
\ No newline at end of file


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/integration/AclProducerAndConsumerIT.java b/src/test/java/com/purbon/kafka/topology/integration/AclProducerAndConsumerIT.java
index b4bee59..1194dd0 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/AclProducerAndConsumerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/AclProducerAndConsumerIT.java
@@ -37,9 +37,7 @@ public final class AclProducerAndConsumerIT {
             .withUser(OTHER_CONSUMER_USERNAME);
     container.start();
     ContainerTestUtils.populateAcls(
-        container,
-        "/acl-producer-and-consumer-it.yaml",
-        "/acl-producer-and-consumer-it.properties");
+        container, "/acl-producer-and-consumer-it.yaml", "/integration-tests.properties");
   }
 
   @AfterClass
diff --git a/src/test/java/com/purbon/kafka/topology/integration/StreamsAclIT.java b/src/test/java/com/purbon/kafka/topology/integration/StreamsAclIT.java
new file mode 100644
index 0000000..8ffddb1
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/integration/StreamsAclIT.java
@@ -0,0 +1,134 @@
+package com.purbon.kafka.topology.integration;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import com.purbon.kafka.topology.integration.containerutils.ContainerTestUtils;
+import com.purbon.kafka.topology.integration.containerutils.SaslPlaintextKafkaContainer;
+import com.purbon.kafka.topology.integration.containerutils.TestConsumer;
+import com.purbon.kafka.topology.integration.containerutils.TestProducer;
+import com.purbon.kafka.topology.integration.containerutils.TestStreams;
+import java.util.Set;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.kstream.KStream;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public final class StreamsAclIT {
+
+  public static final String TOPIC_A = "topic-A";
+  public static final String TOPIC_B = "topic-B";
+  public static final String TOPIC_C = "topic-C";
+  public static final long MAX_TEST_SEC_BEFORE_GIVING_UP = 60;
+  public static final String STREAMS_APP_ID = "streams-appid";
+  private static final String PRODUCER_USERNAME = "producer";
+  private static final String CONSUMER_USERNAME = "consumer";
+  private static final String STREAMS_USERNAME = "streamsapp";
+  private static final String CONSUMER_GROUP = "streams-consumer-test-consumer-group";
+
+  private static SaslPlaintextKafkaContainer container;
+
+  @BeforeClass
+  public static void beforeClass() {
+    container =
+        new SaslPlaintextKafkaContainer()
+            .withUser(PRODUCER_USERNAME)
+            .withUser(CONSUMER_USERNAME)
+            .withUser(STREAMS_USERNAME);
+    container.start();
+    ContainerTestUtils.populateAcls(
+        container, "/streams-acl-it.yaml", "/integration-tests.properties");
+  }
+
+  @AfterClass
+  public static void afterClass() {
+    container.stop();
+  }
+
+  @Test
+  public void shouldNotProduceWithoutPermission() {
+    try (final TestProducer producer = TestProducer.create(container, PRODUCER_USERNAME)) {
+      producer.produceSomeStrings(TOPIC_A);
+    }
+    final StreamsBuilder builder = new StreamsBuilder();
+    KStream<Object, Object> source = builder.stream(TOPIC_A);
+    source.filter((key, val) -> true).to(TOPIC_C);
+
+    final TestStreams streams =
+        TestStreams.create(container, STREAMS_USERNAME, STREAMS_APP_ID, builder.build());
+    streams.start();
+
+    await()
+        .atMost(MAX_TEST_SEC_BEFORE_GIVING_UP, SECONDS)
+        .until(streams::isTopicAuthorizationExceptionThrown);
+
+    streams.close();
+  }
+
+  @Test
+  public void testSimpleStream() {
+    Set<String> values;
+    try (final TestProducer producer = TestProducer.create(container, PRODUCER_USERNAME)) {
+      values = producer.produceSomeStrings(TOPIC_A);
+    }
+
+    final StreamsBuilder builder = new StreamsBuilder();
+    KStream<Object, Object> source = builder.stream(TOPIC_A);
+    source.filter((key, val) -> values.stream().anyMatch(v -> v.equals(val))).to(TOPIC_B);
+
+    final TestStreams streams =
+        TestStreams.create(container, STREAMS_USERNAME, STREAMS_APP_ID, builder.build());
+    streams.start();
+
+    try (final TestConsumer consumer =
+        TestConsumer.create(container, CONSUMER_USERNAME, CONSUMER_GROUP)) {
+      consumer.consumeForAWhile(
+          TOPIC_B,
+          (key, value) -> {
+            values.remove(value);
+            return !values.isEmpty();
+          });
+    }
+    streams.close();
+
+    assertThat(values).isEmpty();
+  }
+
+  @Test
+  public void testStreamWithInternalTopics() {
+    final Set<String> values;
+    try (final TestProducer producer = TestProducer.create(container, PRODUCER_USERNAME)) {
+      values = producer.produceSomeStrings(TOPIC_A);
+    }
+
+    final StreamsBuilder builder = new StreamsBuilder();
+    KStream<String, String> source = builder.stream(TOPIC_A);
+
+    // Just re-group by using value as key, groupBy requires new internal topics
+    source
+        .groupBy((key, value) -> value)
+        .aggregate(() -> "", (aggKey, newValue, aggValue) -> newValue)
+        .filter((key, val) -> values.stream().anyMatch(v -> v.equals(key)))
+        .toStream()
+        .to(TOPIC_B);
+
+    final TestStreams streams =
+        TestStreams.create(container, STREAMS_USERNAME, STREAMS_APP_ID, builder.build());
+    streams.start();
+
+    try (final TestConsumer consumer =
+        TestConsumer.create(container, CONSUMER_USERNAME, CONSUMER_GROUP)) {
+      consumer.consumeForAWhile(
+          TOPIC_B,
+          (key, value) -> {
+            values.remove(key);
+            return !values.isEmpty();
+          });
+    }
+    streams.close();
+
+    assertThat(values).isEmpty();
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestConsumer.java b/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestConsumer.java
index 2f77c8e..170d508 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestConsumer.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestConsumer.java
@@ -15,7 +15,7 @@ public final class TestConsumer implements Closeable {
   /* This timeout includes the time spent fetching meta data, which may actually be several hundred ms.
    * If the timeout is too short, the poll function will return after fetching meta data, but before
    * actually trying to access the topic, thus not throwing the exception we are looking for. */
-  private static final long MAX_MS_TO_CONSUME = 20 * 1000L;
+  private static final long MAX_MS_TO_CONSUME = 60 * 1000L;
   private final KafkaConsumer<String, String> consumer;
 
   public interface RecordHandler {
@@ -54,11 +54,12 @@ public final class TestConsumer implements Closeable {
   public void consumeForAWhile(final String topicName, final RecordHandler handler) {
     consumer.subscribe(Collections.singleton(topicName));
     final long endTime = System.currentTimeMillis() + MAX_MS_TO_CONSUME;
-    while (System.currentTimeMillis() < endTime) {
+    boolean continueConsuming = true;
+    while (continueConsuming && System.currentTimeMillis() < endTime) {
       final ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
       for (final ConsumerRecord<String, String> record : records) {
-        if (handler != null && !handler.handle(record.key(), record.value())) {
-          break;
+        if (handler != null) {
+          continueConsuming = handler.handle(record.key(), record.value());
         }
         consumer.commitAsync();
       }
diff --git a/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestStreams.java b/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestStreams.java
new file mode 100644
index 0000000..b6d5c43
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/integration/containerutils/TestStreams.java
@@ -0,0 +1,71 @@
+package com.purbon.kafka.topology.integration.containerutils;
+
+import java.io.Closeable;
+import java.lang.Thread.UncaughtExceptionHandler;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.kafka.common.errors.TopicAuthorizationException;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+
+public final class TestStreams implements Closeable {
+
+  private final KafkaStreams streams;
+  private boolean topicAuthorizationExceptionThrown;
+
+  private TestStreams(final KafkaStreams streams) {
+    this.streams = streams;
+  }
+
+  public static TestStreams create(
+      final AlternativeKafkaContainer container,
+      final String usernameAndPassword,
+      final String applicationId,
+      Topology topology) {
+    return create(container.getBootstrapServers(), usernameAndPassword, applicationId, topology);
+  }
+
+  private static TestStreams create(
+      final String bootstrapServers,
+      final String usernameAndPassword,
+      final String applicationId,
+      Topology topology) {
+    final Map<String, Object> config =
+        ContainerTestUtils.getSaslConfig(
+            bootstrapServers, usernameAndPassword, usernameAndPassword);
+    config.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+    config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+    config.put(
+        StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+    config.put(StreamsConfig.REQUEST_TIMEOUT_MS_CONFIG, "3000");
+    config.put(StreamsConfig.RETRIES_CONFIG, "0");
+
+    Properties properties = new Properties();
+    properties.putAll(config);
+    return new TestStreams(new KafkaStreams(topology, properties));
+  }
+
+  public void start() {
+    setUncaughtExceptionHandler(
+        (t, e) ->
+            topicAuthorizationExceptionThrown =
+                ExceptionUtils.indexOfType(e, TopicAuthorizationException.class) > 0);
+    streams.start();
+  }
+
+  @Override
+  public void close() {
+    streams.close();
+  }
+
+  public void setUncaughtExceptionHandler(UncaughtExceptionHandler eh) {
+    streams.setUncaughtExceptionHandler(eh);
+  }
+
+  public boolean isTopicAuthorizationExceptionThrown() {
+    return topicAuthorizationExceptionThrown;
+  }
+}
diff --git a/src/test/resources/acl-producer-and-consumer-it.properties b/src/test/resources/integration-tests.properties
similarity index 100%
rename from src/test/resources/acl-producer-and-consumer-it.properties
rename to src/test/resources/integration-tests.properties
diff --git a/src/test/resources/streams-acl-it.yaml b/src/test/resources/streams-acl-it.yaml
new file mode 100644
index 0000000..e3e0e23
--- /dev/null
+++ b/src/test/resources/streams-acl-it.yaml
@@ -0,0 +1,29 @@
+---
+context: "context"
+projects:
+  - name: "streamsproject"
+    streams:
+      - principal: "User:streamsapp"
+        applicationId: "streams-appid"
+        topics:
+          read:
+            - topic-A
+          write:
+            - topic-B
+    producers:
+      - principal: "User:producer"
+    consumers:
+      - principal: "User:consumer"
+    topics:
+      - name: topic-A
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: topic-B
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: topic-C
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
\ No newline at end of file