update readme + examples

Author: idanasulin2706

.idea/runConfigurations/kafka_clients_example.xml

@@ -20,6 +20,7 @@ Works with any Java library that depends on `kafka-clients`, including:
 - Apache Kafka Clients
 - Spring Kafka
 - Alpakka Kafka (Akka Kafka)
+- Pekko Kafka (Apache Pekko)
 - Kafka Streams
 - Kafka Connect
 - Any custom wrapper around the Kafka Java client
@@ -115,6 +116,43 @@ public class KafkaConfig {
 }
 ```
 
+### Pekko/Akka Kafka Applications
+Pekko and Akka Kafka applications typically use immutable configuration maps internally, which prevents Superstream from applying optimizations. To enable Superstream optimizations with Pekko/Akka, you need to create the KafkaProducer manually with a mutable configuration.
+
+❌ **Native Pekko/Akka pattern (optimizations won't be applied)**:
+```java
+ProducerSettings<String, String> producerSettings = ProducerSettings
+    .create(system, new StringSerializer(), new StringSerializer())
+    .withBootstrapServers("localhost:9092");
+
+Source.single(ProducerMessage.single(record))
+    .via(Producer.flexiFlow(producerSettings))
+    .runWith(Sink.ignore, system);
+```
+
+✅ **Superstream-optimized pattern**:
+```java
+// Add these lines to create a mutable producer
+Map<String, Object> configProps = new HashMap<>();
+configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+
+org.apache.kafka.clients.producer.Producer<String, String> kafkaProducer = new KafkaProducer<>(configProps);
+
+ProducerSettings<String, String> producerSettings = ProducerSettings
+    .create(system, new StringSerializer(), new StringSerializer())
+    .withProducer(kafkaProducer);
+
+Source.single(ProducerMessage.single(record))
+    .via(Producer.flexiFlow(producerSettings))
+    .runWith(Sink.ignore, system);
+```
+
+This approach ensures that:
+- All Pekko/Akka Kafka features remain available
+- Superstream can optimize the producer configuration
+
 ### Why This Matters
 The Superstream library needs to modify your producer's configuration to apply optimizations based on your cluster's characteristics. This includes adjusting settings like compression, batch size, and other performance parameters. When the configuration is immutable, these optimizations cannot be applied.
 

README.md

@@ -5,6 +5,8 @@
 import akka.kafka.ProducerSettings;
 import akka.kafka.javadsl.Producer;
 import akka.stream.javadsl.Source;
+import akka.NotUsed;
+import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -15,6 +17,7 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.Properties;
 
 /**
  * Example application that uses Akka Kafka to produce messages.
@@ -47,26 +50,17 @@ public static void main(String[] args) {
         try {
             // Configure the producer
             Map<String, Object> configProps = new HashMap<>();
-            configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
-            configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
-            configProps.put(ProducerConfig.LINGER_MS_CONFIG, 0);
+            configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+            configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+            configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
-            logger.info("Creating Akka Kafka producer with bootstrap servers: {}", bootstrapServers);
-            logger.info("Original producer configuration:");
-            configProps.forEach((k, v) -> logger.info("  {} = {}", k, v));
+            
+            org.apache.kafka.clients.producer.Producer<String, String> kafkaProducer = new KafkaProducer<>(configProps);
 
-            // Create producer settings
-            // Create a new map with String values
-            Map<String, String> stringProps = new HashMap<>();
-            for (Map.Entry<String, Object> entry : configProps.entrySet()) {
-                stringProps.put(entry.getKey(), entry.getValue().toString());
-            }
-
-            // Use the string map with withProperties
+            // Create producer settings with the producer
             ProducerSettings<String, String> producerSettings = ProducerSettings
                     .create(system, new StringSerializer(), new StringSerializer())
-                    .withBootstrapServers(bootstrapServers)
-                    .withProperties(stringProps);
+                    .withProducer(kafkaProducer);
 
             // Send a test message
             String topic = "example-topic";
@@ -80,9 +74,15 @@ public static void main(String[] args) {
             // Create a source with a single message and send it to Kafka
             CompletableFuture<Void> completionFuture = Source.single(ProducerMessage.single(record))
                     .via(Producer.flexiFlow(producerSettings))
-                    .runForeach(result -> logger.info("Message sent: {}", result.toString()), system)
+                    .runForeach(result -> {
+                        if (result instanceof ProducerMessage.Results) {
+                            ProducerMessage.Results<String, String, NotUsed> res = 
+                                (ProducerMessage.Results<String, String, NotUsed>) result;
+                            logger.info("Message sent successfully");
+                        }
+                    }, system)
                     .toCompletableFuture()
-                    .thenApply(done -> null); // Convert CompletableFuture<Done> to CompletableFuture<Void>
+                    .thenApply(done -> null);
 
             // Wait for the message to be sent
             completionFuture.get(10, TimeUnit.SECONDS);

examples/akka-kafka-example/src/main/java/ai/superstream/examples/AkkaKafkaExample.java

@@ -4,7 +4,7 @@
   <groupId>ai.superstream</groupId>
   <artifactId>superstream-clients</artifactId>
   <name>Superstream Kafka Client Optimizer</name>
-  <version>1.0.203</version>
+  <version>1.0.204</version>
   <description>A Java library that dynamically optimizes Kafka client configuration based on recommendations</description>
   <url>https://github.com/superstreamlabs/superstream-clients-java</url>
   <developers>

superstream-clients/dependency-reduced-pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>ai.superstream</groupId>
     <artifactId>superstream-clients</artifactId>
-    <version>1.0.203</version>
+    <version>1.0.204</version>
     <packaging>jar</packaging>
 
     <name>Superstream Kafka Client Optimizer</name>