Testing Kafka Applications

@EmbeddedKafka(partitions = 1, topics = {"orders"})
@SpringBootTest
class OrderTest { /* broker auto-started */ }

var props = KafkaTestUtils.producerProps(embeddedKafka);
var producer = new KafkaProducer<String,String>(props,
  new StringSerializer(), new StringSerializer());
producer.send(new ProducerRecord<>("orders","k","v")).get();

var props = KafkaTestUtils.consumerProps("grp","true",embeddedKafka);
var c = new KafkaConsumer<String,String>(props);
c.subscribe(List.of("orders"));
var rec = KafkaTestUtils.getSingleRecord(c, "orders");
assertEquals("v", rec.value());

var driver = new TopologyTestDriver(topology, props);
var in = driver.createInputTopic("in", ks, vs);
var out = driver.createOutputTopic("out", kd, vd);
in.pipeInput("k", "v");
assertEquals("V", out.readValue());

var mock = new MockProducer<>(true,
  new StringSerializer(), new StringSerializer());
service.publish(mock, "k", "v");
assertEquals(1, mock.history().size());

byte[] bytes = serde.serializer().serialize("t", obj);
Order back = serde.deserializer().deserialize("t", bytes);
assertEquals(obj, back);

var mock = new MockProducer<>(false, ks, vs);
var f = service.publish(mock, "k", "v");
mock.errorNext(new RuntimeException("broker down"));
assertThrows(ExecutionException.class, f::get);

@Container
static KafkaContainer kafka =
  new KafkaContainer(DockerImageName.parse("apache/kafka:3.9.0"));
// kafka.getBootstrapServers() for clients

kafka-producer-perf-test.sh --topic perf --num-records 1000000 \
  --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers=localhost:9092

kafka-producer-perf-test.sh --topic load --num-records 50000000 \
  --record-size 512 --throughput 100000 \
  --producer-props bootstrap.servers=localhost:9092