This extends Apache Kafka example with Java Producer & Consumer Tutorial – Part 2. This covers Apache Kafka example with toxiproxy & producer timeout. The producer code needs to be robust to handle any network issues like timeouts & latency issues. In this tutorial a tool named toxiproxy is used to simulate network issues.
Kafka server on Docker
TOXIPROXY WITH KAFKA
Step 1: As discussed in part-1, stand-up the Kafka, Zookeeper & Apache Kafka Cluster Visualization (AKHQ) on a Docker container.
|
1 2 3 |
docker-compose up -d |
Step 2: Kafka is running on localhost:9093
|
1 2 3 |
$ docker ps |
Outputs:
|
1 2 3 4 5 6 |
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES d469edecca32 tchiotludo/akhq "docker-entrypoint.s…" 5 seconds ago Up 4 seconds (health: starting) 0.0.0.0:9099->8080/tcp simple-kafka_akhq_1 ff69893f064a bitnami/kafka:3.4 "/opt/bitnami/script…" 5 seconds ago Up 5 seconds 9092/tcp, 0.0.0.0:9093->9093/tcp simple-kafka_kafka_1 07e0a44de2a8 bitnami/zookeeper:3.8 "/opt/bitnami/script…" 6 seconds ago Up 5 seconds 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp, 8080/tcp simple-kafka_zookeeper_1 |
Install toxiproxy
Step 3: Install toxiproxy as described. For example, on Mac OS
|
1 2 3 |
$ brew tap shopify/shopify |
and then
|
1 2 3 |
$ brew install toxiproxy |
Run toxiproxy server
Step 4: Open a new command-line & run the toxiproxy server:
|
1 2 3 4 |
$ toxiproxy-server {"level":"info","host":"localhost","port":"8474","version":"2.5.0","caller":"api.go:119","time":"2023-04-16T08:59:28Z","message":"Starting HTTP server on endpoint localhost:8474"} |
toxiproxy cli to create proxies
Step 5: Open a new command-line & run the toxiproxy client see a list of any proxies.
|
1 2 3 |
$ toxiproxy-cli list |
Outputs:
|
1 2 3 4 5 |
Name Listen Upstream Enabled Toxics ====================================================================================== no proxies |
Step 6: Create a proxy for the kafka service running on localhost:9093.
|
1 2 3 |
$ toxiproxy-cli create -l localhost:19093 -u localhost:9093 kafka-proxy |
Outputs:
|
1 2 3 |
Created new proxy kafka-proxy |
You can list the proxies running.
|
1 2 3 |
$ toxiproxy-cli list |
Outputs:
|
1 2 3 4 5 |
Name Listen Upstream Enabled Toxics ====================================================================================== kafka-proxy 127.0.0.1:19093 localhost:9093 enabled None |
Run the ProducerApp.java from Intellij IDEA
Step 7: Please note that the producer connects to the Kafka server running in Docker via the toxiproxy proxy running on localhost:19093 created in the previous step.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
package com.myapp; import org.apache.kafka.clients.producer.*; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Properties; public class ProducerApp { private static final Logger LOG = LoggerFactory.getLogger(ProducerApp.class); public static void main(String[] args) { String topic = "my-first-topic"; Properties properties = new Properties(); properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19093"); properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.setProperty(ProducerConfig.ACKS_CONFIG,"all"); // create the producera KafkaProducer<String, String> producer = new KafkaProducer<>(properties); // create a producer record ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, "Some text to publish"); producer.send(producerRecord, new Callback() { @Override public void onCompletion(RecordMetadata rm, Exception arg1) { System.out.println(arg1); if(arg1!=null) System.out.println(arg1); else System.out.println(rm.topic()+" "+rm.partition()+" "+rm.offset()+" "); } }); // flush data - synchronous producer.flush(); // flush and close producer producer.close(); } } |
If you run the above code the “ProducerApp.java” will successfully publish a message to the topic “my-first-topic“. You can view the message in the UI via a browser with URL http://localhost:9099/ui/docker-kafka-server/topic.
Note that below lines are added to acknowledge if a message can’t be published. This gets used in the later steps when a timeout toxic is added to simulate a network timeout issue with toxiproxy.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
properties.setProperty(ProducerConfig.ACKS_CONFIG,"all"); ..... producer.send(producerRecord, new Callback() { @Override public void onCompletion(RecordMetadata rm, Exception arg1) { System.out.println(arg1); if(arg1!=null) System.out.println(arg1); else System.out.println(rm.topic()+" "+rm.partition()+" "+rm.offset()+" "); } }); |
Add timeout toxic to the kafka-proxy
Step 8: Let’s add a timeout toxic to the proxy kafka-proxy that was created via toxiproxy-cli in one of the previous steps.
|
1 2 3 |
toxiproxy-cli toxic add --type timeout kafka-proxy |
Outputs:
|
1 2 3 |
Added downstream timeout toxic 'timeout_downstream' on proxy 'kafka-proxy' |
You can list the toxics with inspect.
|
1 2 3 |
toxiproxy-cli inspect kafka-proxy |
Outputs:
|
1 2 3 4 5 6 7 8 9 |
Name: kafka-proxy Listen: 127.0.0.1:19093 Upstream: localhost:9093 ====================================================================== Upstream toxics: Proxy has no Upstream toxics enabled. Downstream toxics: timeout_downstream: type=timeout stream=downstream toxicity=1.00 attributes=[ timeout=0 ] |
Step 9: Run the Java application “ProducerApp.java” again to connect to the Kafka server at “localhost:19093” with a timeout toxic. You will notice that the “ProducerApp.java” will be running for a while & gets timed out.
It gets timed out on the line
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
.... producer.send(producerRecord, new Callback() { @Override public void onCompletion(RecordMetadata rm, Exception arg1) { System.out.println(arg1); if(arg1!=null) System.out.println(arg1); else System.out.println(rm.topic()+" "+rm.partition()+" "+rm.offset()+" "); } }); ... |
Outputs:
|
1 2 3 4 5 |
org.apache.kafka.common.errors.TimeoutException: Topic my-first-topic not present in metadata after 60000 ms. org.apache.kafka.common.errors.TimeoutException: Topic my-first-topic not present in metadata after 60000 ms. Exits the app. |
Remove the toxic whilst the ProducerApp.java is running
Step 9: Run the Java application “ProducerApp.java” again to connect to the Kafka server at “localhost:19093” with a timeout toxic.
Whilst the application is running & blocked on producer.send(..,..), remove the toxic with the below command-line.
|
1 2 3 |
$ toxiproxy-cli toxic remove -n timeout_downstream kafka-proxy |
Now the message will be published to the Kafka topic “my-first-topic“.
|
1 2 3 4 |
null my-first-topic 0 2 |
Add latency toxic
Step 10: Add a latency toxic as shown below:
|
1 2 3 |
$ toxiproxy-cli toxic add --type latency --attribute latency=2000 kafka-proxy |
Outputs:
|
1 2 3 |
Added downstream latency toxic 'latency_downstream' on proxy 'kafka-proxy' |
List the toxics:
|
1 2 3 |
$ toxiproxy-cli inspect kafka-proxy |
Outputs:
|
1 2 3 4 5 6 7 8 9 |
Name: kafka-proxy Listen: 127.0.0.1:19093 Upstream: localhost:9093 ====================================================================== Upstream toxics: Proxy has no Upstream toxics enabled. Downstream toxics: latency_downstream: type=latency stream=downstream toxicity=1.00 attributes=[ jitter=0 latency=2000 ] |
Step 11: Run the Java application “ProducerApp.java” again to connect to the Kafka server at “localhost:19093” with a latency toxic.
The code will take extra 2 seconds, but the message will be published to the topic.
|
1 2 3 4 |
null my-first-topic 0 3 |
Finally, remove the proxy
|
1 2 3 |
$ toxiproxy-cli delete kafka-proxy |
Outputs:
|
1 2 3 |
Deleted proxy kafka-proxy |