wip

6 files changed, 165 insertions, 0 deletions

diff --git a/.gitignore b/.gitignore
index 9fb18b4..cfaa919 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,5 @@
 .idea
 out
+*.tmp
+tmp
+*.tmp.*
diff --git a/kafka-sandbox/.gitignore b/kafka-sandbox/.gitignore
new file mode 100644
index 0000000..5ff6309
--- /dev/null
+++ b/kafka-sandbox/.gitignore
@@ -0,0 +1,38 @@
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### IntelliJ IDEA ###
+.idea/modules.xml
+.idea/jarRepositories.xml
+.idea/compiler.xml
+.idea/libraries/
+*.iws
+*.iml
+*.ipr
+
+### Eclipse ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/
+
+### Mac OS ###
+.DS_Store
\ No newline at end of file
diff --git a/kafka-sandbox/kafka.properties b/kafka-sandbox/kafka.properties
new file mode 100644
index 0000000..45042ad
--- /dev/null
+++ b/kafka-sandbox/kafka.properties
@@ -0,0 +1,5 @@
+bootstrap.servers=localhost:9092
+key.serializer=org.apache.kafka.common.serialization.StringSerializer
+value.serializer=org.apache.kafka.common.serialization.StringSerializer
+key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
diff --git a/kafka-sandbox/pom.xml b/kafka-sandbox/pom.xml
new file mode 100644
index 0000000..eb33e8d
--- /dev/null
+++ b/kafka-sandbox/pom.xml
@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>io.trygvis</groupId>
+  <artifactId>kafka-sandbox</artifactId>
+  <version>1.0-SNAPSHOT</version>
+
+  <properties>
+    <maven.compiler.source>21</maven.compiler.source>
+    <maven.compiler.target>21</maven.compiler.target>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    <kafka-clients.version>3.5.1</kafka-clients.version>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>${kafka-clients.version}</version>
+    </dependency>
+  </dependencies>
+
+</project>
diff --git a/kafka-sandbox/src/main/java/examples/ConsumerExample.java b/kafka-sandbox/src/main/java/examples/ConsumerExample.java
new file mode 100644
index 0000000..e491035
--- /dev/null
+++ b/kafka-sandbox/src/main/java/examples/ConsumerExample.java
@@ -0,0 +1,38 @@
+package examples;
+
+import org.apache.kafka.clients.consumer.*;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Properties;
+
+public class ConsumerExample {
+    public static void main(final String[] args) throws Exception {
+        if (args.length != 1) {
+            System.out.println("Please provide the configuration file path as a command line argument");
+            System.exit(1);
+        }
+
+        final String topic = "purchases";
+
+        // Load consumer configuration settings from a local file
+        // Reusing the loadConfig method from the ProducerExample class
+        final Properties props = ProducerExample.loadConfig(args[0]);
+
+        // Add additional properties.
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka-java-getting-started");
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+        try (final Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
+            consumer.subscribe(Arrays.asList(topic));
+            while (true) {
+                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
+                for (ConsumerRecord<String, String> record : records) {
+                    String key = record.key();
+                    String value = record.value();
+                    System.out.println(String.format("Consumed event from topic %s: key = %-10s value = %s", topic, key, value));
+                }
+            }
+        }
+    }
+}
diff --git a/kafka-sandbox/src/main/java/examples/ProducerExample.java b/kafka-sandbox/src/main/java/examples/ProducerExample.java
new file mode 100644
index 0000000..3d185cd
--- /dev/null
+++ b/kafka-sandbox/src/main/java/examples/ProducerExample.java
@@ -0,0 +1,55 @@
+package examples;
+
+import org.apache.kafka.clients.producer.*;
+
+import java.io.*;
+import java.nio.file.*;
+import java.util.*;
+
+public class ProducerExample {
+
+    public static void main(final String[] args) throws IOException {
+        if (args.length != 1) {
+            System.out.println("Please provide the configuration file path as a command line argument");
+            System.exit(1);
+        }
+
+        // Load producer configuration settings from a local file
+        final Properties props = loadConfig(args[0]);
+        final String topic = "purchases";
+
+        String[] users = {"eabara", "jsmith", "sgarcia", "jbernard", "htanaka", "awalther"};
+        String[] items = {"book", "alarm clock", "t-shirts", "gift card", "batteries"};
+        try (final Producer<String, String> producer = new KafkaProducer<>(props)) {
+            final Random rnd = new Random();
+            final var numMessages = 10L;
+            for (var i = 0L; i < numMessages; i++) {
+                String user = users[rnd.nextInt(users.length)];
+                String item = items[rnd.nextInt(items.length)];
+
+                producer.send(
+                        new ProducerRecord<>(topic, user, item),
+                        (event, ex) -> {
+                            if (ex != null)
+                                ex.printStackTrace();
+                            else
+                                System.out.printf("Produced event to topic %s: key = %-10s value = %s%n", topic, user, item);
+                        });
+            }
+            System.out.printf("%s events were produced to topic %s%n", numMessages, topic);
+        }
+
+    }
+
+    // We'll reuse this function to load properties from the Consumer as well
+    public static Properties loadConfig(final String configFile) throws IOException {
+        if (!Files.exists(Paths.get(configFile))) {
+            throw new IOException(configFile + " not found.");
+        }
+        final Properties cfg = new Properties();
+        try (InputStream inputStream = new FileInputStream(configFile)) {
+            cfg.load(inputStream);
+        }
+        return cfg;
+    }
+}