null checks + examples and readme fixes

Author: idanasulin2706

.idea/runConfigurations/kafka_clients_example.xml

@@ -31,6 +31,97 @@ Works with any Java library that depends on `kafka-clients`, including:
 - **Intelligent optimization**: Identifies the most impactful topics to optimize
 - **Graceful fallback**: Falls back to default settings if optimization fails
 
+## Important: Producer Configuration Requirements
+
+When initializing your Kafka producers, please ensure you pass the configuration as a mutable object. The Superstream library needs to modify the producer configuration to apply optimizations. The following initialization patterns are supported:
+
+✅ **Supported (Recommended)**:
+```java
+// Using Properties (recommended)
+Properties props = new Properties();
+props.put("bootstrap.servers", "localhost:9092");
+// ... other properties ...
+KafkaProducer<String, String> producer = new KafkaProducer<>(props);
+
+// Using a regular HashMap
+Map<String, Object> config = new HashMap<>();
+config.put("bootstrap.servers", "localhost:9092");
+// ... other properties ...
+KafkaProducer<String, String> producer = new KafkaProducer<>(config);
+
+// Using Spring's @Value annotations and configuration loading
+@Configuration
+public class KafkaConfig {
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+    // ... other properties ...
+
+    @Bean
+    public ProducerFactory<String, String> producerFactory() {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        // ... other properties ...
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+}
+```
+
+❌ **Not Supported**:
+```java
+// Using Collections.unmodifiableMap
+Map<String, Object> config = Collections.unmodifiableMap(new HashMap<>());
+KafkaProducer<String, String> producer = new KafkaProducer<>(config);
+
+// Using Map.of() (creates unmodifiable map)
+KafkaProducer<String, String> producer = new KafkaProducer<>(
+    Map.of("bootstrap.servers", "localhost:9092")
+);
+
+// Using ProducerConfig.originals() which returns an unmodifiable copy
+ProducerConfig config = new ProducerConfig(props);
+KafkaProducer<String, String> producer = new KafkaProducer<>(config.originals());
+
+// Using KafkaTemplate's getProducerFactory().getConfigurationProperties()
+// which returns an unmodifiable map
+KafkaTemplate<String, String> template = new KafkaTemplate<>(producerFactory);
+KafkaProducer<String, String> producer = new KafkaProducer<>(
+    template.getProducerFactory().getConfigurationProperties()
+);
+```
+
+### Spring Applications
+Spring applications that use `@Value` annotations and Spring's configuration loading (like `application.yml` or `application.properties`) are fully supported. The Superstream library will be able to modify the configuration when it's loaded into a mutable `Map` or `Properties` object in your Spring configuration class.
+
+Example of supported Spring configuration:
+```yaml
+# application.yml
+spring:
+  kafka:
+    producer:
+      properties:
+        compression.type: snappy
+        batch.size: 16384
+        linger.ms: 1
+```
+
+```java
+@Configuration
+public class KafkaConfig {
+    @Value("${spring.kafka.producer.properties.compression.type}")
+    private String compressionType;
+    
+    @Bean
+    public ProducerFactory<String, String> producerFactory() {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType);
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+}
+```
+
+### 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.
+
 ## Installation
 
 *Superstream package*: https://central.sonatype.com/artifact/ai.superstream/superstream-clients-java/overview

.idea/runConfigurations/spring_example.xml

@@ -0,0 +1,56 @@
+package ai.superstream;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfig {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    @Value("${spring.kafka.producer.key-serializer}")
+    private String keySerializer;
+
+    @Value("${spring.kafka.producer.value-serializer}")
+    private String valueSerializer;
+
+    @Value("${spring.kafka.producer.client-id}")
+    private String clientId;
+
+    @Value("${spring.kafka.producer.properties.compression.type}")
+    private String compressionType;
+
+    @Value("${spring.kafka.producer.properties.batch.size}")
+    private String batchSize;
+
+    @Value("${spring.kafka.producer.properties.linger.ms}")
+    private String lingerMs;
+
+    @Bean
+    public ProducerFactory<String, String> producerFactory() {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
+        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
+        configProps.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
+        configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType);
+        configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
+        configProps.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
+
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> kafkaTemplate() {
+        return new KafkaTemplate<>(producerFactory());
+    }
+} 

README.md

@@ -0,0 +1,26 @@
+package ai.superstream;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+public class KafkaProducerService {
+
+    private final KafkaTemplate<String, String> kafkaTemplate;
+    
+    @Value("${kafka.topic}")
+    private String topic;
+
+    public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
+        this.kafkaTemplate = kafkaTemplate;
+    }
+
+    public void sendMessage(String message) {
+        kafkaTemplate.send(topic, message);
+    }
+
+    public void sendMessageWithKey(String key, String message) {
+        kafkaTemplate.send(topic, key, message);
+    }
+} 

examples/spring-kafka-example/src/main/java/ai/superstream/KafkaProducerConfig.java

@@ -0,0 +1,26 @@
+package ai.superstream;
+
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.stereotype.Component;
+
+@Component
+public class MessageSender implements CommandLineRunner {
+
+    private final KafkaProducerService producerService;
+
+    public MessageSender(KafkaProducerService producerService) {
+        this.producerService = producerService;
+    }
+
+    @Override
+    public void run(String... args) {
+        // Send a simple message
+        producerService.sendMessage("Hello from Spring Kafka Example!");
+        
+        // Send a message with a key
+        producerService.sendMessageWithKey("user-1", "Message for user 1");
+        producerService.sendMessageWithKey("user-2", "Message for user 2");
+        
+        System.out.println("Messages sent successfully!");
+    }
+} 

examples/spring-kafka-example/src/main/java/ai/superstream/KafkaProducerService.java

@@ -0,0 +1,11 @@
+package ai.superstream;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class SpringKafkaExample {
+    public static void main(String[] args) {
+        SpringApplication.run(SpringKafkaExample.class, args);
+    }
+}