Add docs

streams-bootstrap-test/src/main/java/com/bakdata/kafka/ReaderBuilder.java

@@ -89,9 +89,19 @@ public ReaderBuilder with(final String key, final Object value) {
      * @param <V> type of values
      */
     public <K, V> List<ConsumerRecord<K, V>> from(final String topic, final Duration timeout) {
-        try (final Consumer<K, V> consumer = new KafkaConsumer<>(this.properties)) {
+        try (final Consumer<K, V> consumer = this.createConsumer()) {
             return readAll(consumer, topic, timeout);
         }
     }
 
+    /**
+     * Create a new {@code Consumer} for a Kafka cluster
+     * @return {@code Consumer}
+     * @param <K> type of keys
+     * @param <V> type of values
+     */
+    public <K, V> Consumer<K, V> createConsumer() {
+        return new KafkaConsumer<>(this.properties);
+    }
+
 }

streams-bootstrap-test/src/main/java/com/bakdata/kafka/SenderBuilder.java

@@ -63,11 +63,21 @@ public SenderBuilder with(final String key, final Object value) {
      * @param <V> type of values
      */
     public <K, V> void to(final String topic, final Iterable<SimpleProducerRecord<K, V>> records) {
-        try (final Producer<K, V> producer = new KafkaProducer<>(this.properties)) {
+        try (final Producer<K, V> producer = this.createProducer()) {
             records.forEach(kv -> producer.send(kv.toProducerRecord(topic)));
         }
     }
 
+    /**
+     * Create a new {@code Producer} for a Kafka cluster
+     * @return {@code Producer}
+     * @param <K> type of keys
+     * @param <V> type of values
+     */
+    public <K, V> Producer<K, V> createProducer() {
+        return new KafkaProducer<>(this.properties);
+    }
+
     /**
      * Represents a {@link ProducerRecord} without topic assignment
      * @param <K> type of keys