implement the two dump classes and add mvn profiles

payments-streams-processor/pom.xml

@@ -146,5 +146,61 @@
                 <quarkus.native.enabled>true</quarkus.native.enabled>
             </properties>
         </profile>
+        <!-- dumps current account data -->
+        <profile>
+            <id>dump-accounts</id>
+            <activation>
+                <property>
+                    <name>dump-accounts</name>
+                </property>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.codehaus.mojo</groupId>
+                        <artifactId>exec-maven-plugin</artifactId>
+                        <version>3.4.1</version>
+                        <executions>
+                            <execution>
+                                <goals>
+                                    <goal>java</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                        <configuration>
+                            <mainClass>com.redhat.training.kafka.coreapi.streams.GetCurrentBalance</mainClass>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+        <!-- dumps current risk assessment data -->
+        <profile>
+            <id>dump-risk</id>
+            <activation>
+                <property>
+                    <name>dump-risk</name>
+                </property>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.codehaus.mojo</groupId>
+                        <artifactId>exec-maven-plugin</artifactId>
+                        <version>3.4.1</version>
+                        <executions>
+                            <execution>
+                                <goals>
+                                    <goal>java</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                        <configuration>
+                            <mainClass>com.redhat.training.kafka.coreapi.streams.GetCurrentRiskStatus</mainClass>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
     </profiles>
 </project>

payments-streams-processor/src/main/java/com/redhat/training/kafka/coreapi/streams/GetCurrentBalance.java

@@ -1,5 +1,73 @@
 package com.redhat.training.kafka.coreapi.streams;
 
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.eclipse.microprofile.config.Config;
+import org.eclipse.microprofile.config.ConfigProvider;
+
+import com.redhat.training.kafka.model.BankAccount;
+
+import io.quarkus.kafka.client.serialization.ObjectMapperSerde;
+
+// Just dumps the latest contents of the account-data KTable.
 public class GetCurrentBalance {
-    // TODO: just dump the contents of the account-data KTable
+    public static Properties configureProperties() {
+        Config cf = ConfigProvider.getConfig();
+
+        Properties props = new Properties();
+
+        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "currentBalanceDump");
+        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cf.getValue("kafka.server", String.class));
+        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, cf.getOptionalValue("kafka.protocol", String.class).orElse("PLAINTEXT"));
+        if (props.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG).equals("SSL")) {
+            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, cf.getValue("ssl.truststore", String.class));
+            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, cf.getValue("ssl.password", String.class));
+        }
+
+        return props;
+    }
+
+    public static void main(String[] args) {
+        Properties cfg = configureProperties();
+        String srcTopic = ConfigProvider.getConfig().getOptionalValue("kafka.topic.accounts", String.class).orElse("account-data");
+
+        Serde<String> ks = Serdes.String();
+        Serde<BankAccount> vs = new ObjectMapperSerde<>(BankAccount.class);
+
+        StreamsBuilder b = new StreamsBuilder();
+
+        b.table(srcTopic, Consumed.with(ks, vs)).toStream().foreach((acctId, acct) -> {
+            System.out.println(acct.getAccountNumber() + ": " + acct.getCustomerName() + "(" + acct.getCustomerId() + ")\t-> " + acct.getBalance());
+        });
+        Topology t = b.build();
+        KafkaStreams str = new KafkaStreams(t, cfg);
+
+        final CountDownLatch cd = new CountDownLatch(1);
+        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown") {
+            @Override
+            public void run() {
+                str.close();
+                cd.countDown();
+            }
+        });
+
+        // start the application
+        try {
+            System.out.println("Dumping current account balances...");
+            str.start();
+            cd.await();
+        } catch (InterruptedException ie) {
+            System.out.println("Interrupted during await()...");
+        }
+    }
 }

payments-streams-processor/src/main/java/com/redhat/training/kafka/coreapi/streams/GetCurrentRiskStatus.java

@@ -1,5 +1,74 @@
 package com.redhat.training.kafka.coreapi.streams;
 
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.eclipse.microprofile.config.Config;
+import org.eclipse.microprofile.config.ConfigProvider;
+
+import com.redhat.training.kafka.model.BankAccount;
+import com.redhat.training.kafka.model.RiskAssessment;
+
+import io.quarkus.kafka.client.serialization.ObjectMapperSerde;
+
+// Just dump the current contents of customer-risk-status KTable.
 public class GetCurrentRiskStatus {
-    // TODO: just dump the contents of customer-risk-status KTable
+    public static Properties configureProperties() {
+        Config cf = ConfigProvider.getConfig();
+
+        Properties props = new Properties();
+
+        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "currentBalanceDump");
+        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cf.getValue("kafka.server", String.class));
+        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, cf.getOptionalValue("kafka.protocol", String.class).orElse("PLAINTEXT"));
+        if (props.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG).equals("SSL")) {
+            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, cf.getValue("ssl.truststore", String.class));
+            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, cf.getValue("ssl.password", String.class));
+        }
+
+        return props;
+    }
+
+    public static void main(String[] args) {
+        Properties cfg = configureProperties();
+        String srcTopic = ConfigProvider.getConfig().getOptionalValue("kafka.topic.risk-status", String.class).orElse("customer-risk-status");
+
+        Serde<String> ks = Serdes.String();
+        Serde<RiskAssessment> vs = new ObjectMapperSerde<>(RiskAssessment.class);
+
+        StreamsBuilder b = new StreamsBuilder();
+
+        b.table(srcTopic, Consumed.with(ks, vs)).toStream().foreach((cId, risk) -> {
+            System.out.println(risk.getCustomerId() + " -> " + risk.getAssessmentScore());
+        });
+        Topology t = b.build();
+        KafkaStreams str = new KafkaStreams(t, cfg);
+
+        final CountDownLatch cd = new CountDownLatch(1);
+        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown") {
+            @Override
+            public void run() {
+                str.close();
+                cd.countDown();
+            }
+        });
+
+        // start the application
+        try {
+            System.out.println("Dumping current risk status...");
+            str.start();
+            cd.await();
+        } catch (InterruptedException ie) {
+            System.out.println("Interrupted during await()...");
+        }
+    }
 }

payments-streams-processor/src/main/resources/META-INF/microprofile-config.properties

@@ -0,0 +1,6 @@
+kafka.server = localhost:9092,localhost:9192,localhost:9292
+# kafka.protocol = PLAINTEXT
+# ssl.truststore = path/to/truststore.jks
+# ssl.password = truststorepass
+# kafka.topic.accounts =
+# kafka.topic.risk-status =