Merge pull request #46 from superstreamlabs/master

Release

Author: idanasulin2706

Jenkinsfile

@@ -2,11 +2,13 @@ def isManualTrigger() {
     return currentBuild.getBuildCauses().any { cause -> cause._class == 'hudson.model.Cause$UserIdCause' }
 }
 
+@Library('shared-library') _
+
 pipeline {
 
     agent {
         docker {
-            label 'memphis-jenkins-big-fleet,'
+            label 'big-ec2-fleet'
             image 'maven:3.8.4-openjdk-17'
             args '-u root'
         }
@@ -19,6 +21,17 @@ pipeline {
 
     stages {
         stage('Read Version from pom.xml') {  
+            when {
+                anyOf {
+                    allOf {
+                        branch 'master'
+                        triggeredBy 'UserIdCause' // Manual trigger on master
+                    }
+                    allOf {
+                        branch 'latest'
+                    }
+                }
+            }            
             steps {
                 dir('superstream-clients'){
                     script {
@@ -120,6 +133,48 @@ pipeline {
         always {
             cleanWs()
         }
+        success {
+            script {
+                if (env.GIT_BRANCH == 'latest') {   
+                    sendSlackNotification('SUCCESS')         
+                    notifySuccessful()
+                }
+            }
+        }
+        
+        failure {
+            script {
+                if (env.GIT_BRANCH == 'latest') { 
+                    sendSlackNotification('FAILURE')              
+                    notifyFailed()
+                }
+            }            
+        }
+        aborted {
+            script {
+                if (env.BRANCH_NAME == 'latest') {
+                    sendSlackNotification('ABORTED')
+                }
+                // Get the build log to check for the specific exception and retry job
+                AgentOfflineException()
+            }          
+        }        
     }    
 }
 
+def notifySuccessful() {
+    emailext (
+        subject: "SUCCESSFUL: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]'",
+        body: """SUCCESSFUL: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]':
+        Check console output and connection attributes at ${env.BUILD_URL}""",
+        to: '[email protected]'
+    )
+}
+def notifyFailed() {
+    emailext (
+        subject: "FAILED: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]'",
+        body: """FAILED: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]':
+        Check console output at ${env.BUILD_URL}""",
+        to: '[email protected]'
+    )
+}

README.md

@@ -36,6 +36,20 @@ Works with any Java library that depends on `kafka-clients`, including:
 
 The library fully supports Java versions 11 through 21.
 
+## Run Modes
+
+Superstream operates in one of two modes per `KafkaProducer` instance:
+
+- **INIT_CONFIG**: Superstream applies init-time configuration optimizations directly to the application's producer configuration before the producer is constructed. No shadow producer is created. The application calls run on the original producer.
+- **SHADOW**: Superstream initializes a shadow producer effectively replacing original producer and routes supported producer methods through it. After a learning period, a one-shot optimized configuration is applied to the shadow. The original producer remains in place; routing is transparent.
+
+### Default Optimizations
+
+When no topic-specific configuration is available (or during init-time defaults), Superstream applies:
+- `compression.type=zstd`
+- `batch.size=65536`
+- `linger.ms=5000` (skipped when `SUPERSTREAM_LATENCY_SENSITIVE=true`)
+
 ## 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:
@@ -200,10 +214,6 @@ ENTRYPOINT ["java", "-javaagent:/app/superstream-agent.jar", "-jar", "/app/app.j
 Alternatively, you can use a multi-stage build to download the agent from Maven Central:
 
 
-### Required Environment Variables
-
-- `SUPERSTREAM_TOPICS_LIST`: Comma-separated list of topics your application produces to
-
 ### Optional Environment Variables
 
 - `SUPERSTREAM_LATENCY_SENSITIVE`: Set to "true" to prevent any modification to linger.ms values
@@ -212,7 +222,6 @@ Alternatively, you can use a multi-stage build to download the agent from Maven
 
 Example:
 ```bash
-export SUPERSTREAM_TOPICS_LIST=orders,payments,user-events
 export SUPERSTREAM_LATENCY_SENSITIVE=true
 ```
 

examples/kafka-clients-example/src/main/java/ai/superstream/examples/shadow/AbortTransactionExample.java

@@ -0,0 +1,35 @@
+package ai.superstream.examples.shadow;
+
+import java.time.Duration;
+import java.util.Properties;
+
+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;
+
+public class AbortTransactionExample {
+	public static void main(String[] args) throws Exception {
+		String bootstrap = System.getenv().getOrDefault("BOOTSTRAP_SERVERS", "localhost:9092");
+		String topic = System.getenv().getOrDefault("TOPIC", "example-topic");
+
+		Properties p = new Properties();
+		p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+		p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+		p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+		p.put(ProducerConfig.CLIENT_ID_CONFIG, "shadow-abort");
+		p.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "abort-" + System.currentTimeMillis());
+
+		KafkaProducer<String, String> producer = new KafkaProducer<>(p);
+		try {
+			producer.initTransactions();
+			producer.beginTransaction();
+			producer.send(new ProducerRecord<>(topic, "abort-key", "abort-value"));
+			producer.abortTransaction();
+		} finally {
+			producer.close(Duration.ofSeconds(1));
+		}
+	}
+}
+
+

examples/kafka-clients-example/src/main/java/ai/superstream/examples/shadow/AsyncCallbackExample.java

@@ -0,0 +1,50 @@
+package ai.superstream.examples.shadow;
+
+import java.time.Duration;
+import java.util.Properties;
+
+import org.apache.kafka.clients.producer.Callback;
+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.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AsyncCallbackExample {
+	private static final Logger LOG = LoggerFactory.getLogger(AsyncCallbackExample.class);
+
+	public static void main(String[] args) throws Exception {
+		String bootstrap = System.getenv().getOrDefault("BOOTSTRAP_SERVERS", "localhost:9092");
+		String topic = System.getenv().getOrDefault("TOPIC", "example-topic");
+
+		Properties p = new Properties();
+		p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+		p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+		p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+		p.put(ProducerConfig.CLIENT_ID_CONFIG, "shadow-async-callback");
+
+		KafkaProducer<String, String> producer = new KafkaProducer<>(p);
+		try {
+			producer.send(new ProducerRecord<>(topic, "async-key", "async-value"), new Callback() {
+				@Override
+				public void onCompletion(RecordMetadata metadata, Exception exception) {
+					if (exception != null) {
+						LOG.warn("async send error", exception);
+					} else if (metadata != null) {
+						LOG.info("async send success: topic={}, partition={}, offset={}",
+								metadata.topic(), metadata.partition(), metadata.offset());
+					} else {
+						LOG.info("async send completed without metadata");
+					}
+				}
+			});
+			producer.flush();
+		} finally {
+			producer.close(Duration.ofSeconds(1));
+		}
+	}
+}
+
+

examples/kafka-clients-example/src/main/java/ai/superstream/examples/shadow/CloseOverloadsExample.java

@@ -0,0 +1,34 @@
+package ai.superstream.examples.shadow;
+
+import java.time.Duration;
+import java.util.Properties;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CloseOverloadsExample {
+    private static final Logger LOG = LoggerFactory.getLogger(CloseOverloadsExample.class);
+
+    public static void main(String[] args) throws Exception {
+        String bootstrap = System.getenv().getOrDefault("BOOTSTRAP_SERVERS", "localhost:9092");
+
+        Properties p = new Properties();
+        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        p.put(ProducerConfig.CLIENT_ID_CONFIG, "shadow-close-overloads");
+
+        KafkaProducer<String, String> producer = new KafkaProducer<>(p);
+        try {
+            producer.close(Duration.ofMillis(1000));
+            LOG.info("Invoked close(Duration)");
+        } finally {
+            try { producer.close(); } catch (Throwable ignored) { } // shadow should log that it already closed
+        }
+    }
+}
+
+

examples/kafka-clients-example/src/main/java/ai/superstream/examples/shadow/CustomStringSerializer.java

@@ -0,0 +1,26 @@
+package ai.superstream.examples.shadow;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+
+import org.apache.kafka.common.serialization.Serializer;
+
+public class CustomStringSerializer implements Serializer<String> {
+    @Override
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        // no-op
+    }
+
+    @Override
+    public byte[] serialize(String topic, String data) {
+        if (data == null) return null;
+        return data.getBytes(StandardCharsets.UTF_8);
+    }
+
+    @Override
+    public void close() {
+        // no-op
+    }
+}
+
+

examples/kafka-clients-example/src/main/java/ai/superstream/examples/shadow/HotSwapExample.java

@@ -0,0 +1,43 @@
+package ai.superstream.examples.shadow;
+
+import java.time.Duration;
+import java.util.Properties;
+
+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;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class HotSwapExample {
+	private static final Logger LOG = LoggerFactory.getLogger(HotSwapExample.class);
+
+	public static void main(String[] args) throws Exception {
+		String bootstrap = System.getenv().getOrDefault("BOOTSTRAP_SERVERS", "localhost:9092");
+		String topic = System.getenv().getOrDefault("TOPIC", "example-topic");
+
+		Properties p = new Properties();
+		p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+		p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+		p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+		p.put(ProducerConfig.CLIENT_ID_CONFIG, "shadow-hot-swap");
+
+		KafkaProducer<String, String> producer = new KafkaProducer<>(p);
+		Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+			try {
+				LOG.info("Shutdown: flushing and closing producer");
+				producer.flush();
+				producer.close(Duration.ofSeconds(1));
+			} catch (Throwable ignored) { }
+		}));
+
+		long i = 0;
+		while (true) {
+			producer.send(new ProducerRecord<>(topic, "loop-key", "loop-" + (i++)));
+			Thread.sleep(5_000L);
+		}
+	}
+}
+
+

examples/kafka-clients-example/src/main/java/ai/superstream/examples/shadow/HundredProducersSendExample.java

@@ -0,0 +1,61 @@
+package ai.superstream.examples.shadow;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+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;
+
+public class HundredProducersSendExample {
+	public static void main(String[] args) throws Exception {
+		String bootstrap = System.getenv().getOrDefault("BOOTSTRAP_SERVERS", "localhost:9092");
+		String topic = System.getenv().getOrDefault("TOPIC", "example-topic");
+
+		List<KafkaProducer<String, String>> producers = new ArrayList<>();
+		try {
+			// Create 100 producers
+			for (int i = 0; i < 100; i++) {
+				Properties p = new Properties();
+				p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+				p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+				p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+				p.put(ProducerConfig.CLIENT_ID_CONFIG, "shadow-100-" + i);
+				KafkaProducer<String, String> producer = new KafkaProducer<>(p);
+				producers.add(producer);
+			}
+
+			long startNs = System.nanoTime();
+			int ticks = 12; // every 10s for 2 minutes total
+			for (int t = 0; t < ticks; t++) {
+				// Each producer sends once per tick
+				for (int i = 0; i < producers.size(); i++) {
+					producers.get(i).send(new ProducerRecord<>(topic, "many-" + i, "value-t" + t));
+				}
+				// Flush all producers after the wave
+				for (KafkaProducer<String, String> producer : producers) {
+					producer.flush();
+				}
+				// Sleep 10s between ticks except after the last wave
+				if (t < ticks - 1) {
+					Thread.sleep(10_000L);
+				}
+			}
+			// Ensure total runtime ~2 minutes
+			long elapsedMs = (System.nanoTime() - startNs) / 1_000_000L;
+			long remaining = 120_000L - elapsedMs;
+			if (remaining > 0) {
+				Thread.sleep(remaining);
+			}
+		} finally {
+			for (KafkaProducer<String, String> producer : producers) {
+				try { producer.close(Duration.ofSeconds(1)); } catch (Throwable ignored) { }
+			}
+		}
+	}
+}
+
+