+++ /dev/null
-#!/bin/bash
-
-IMAGE=juplo/spring-producer:1.0-json-messages-SNAPSHOT
-
-if [ "$1" = "cleanup" ]
-then
- docker compose -f docker/docker-compose.yml down -t0 -v --remove-orphans
- mvn clean
- exit
-fi
-
-docker compose -f docker/docker-compose.yml up -d --remove-orphans kafka-1 kafka-2 kafka-3
-docker compose -f docker/docker-compose.yml rm -svf producer
-
-if [[
- $(docker image ls -q $IMAGE) == "" ||
- "$1" = "build"
-]]
-then
- mvn clean install || exit
-else
- echo "Using image existing images:"
- docker image ls $IMAGE
-fi
-
-docker compose -f docker/docker-compose.yml up --remove-orphans setup || exit 1
-
-
-docker compose -f docker/docker-compose.yml up -d producer
-docker compose -f docker/docker-compose.yml up -d peter ute
-sleep 15
-
-docker compose -f docker/docker-compose.yml stop producer
-
-echo
-echo "Von peter empfangen:"
-docker compose -f docker/docker-compose.yml logs peter | grep '\ test\/.'
-echo
-echo "Von ute empfangen:"
-docker compose -f docker/docker-compose.yml logs ute | grep '\ test\/.'
-
-docker compose -f docker/docker-compose.yml stop peter ute
package de.juplo.kafka;
-import de.juplo.messages.Message;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
public ExampleProducer exampleProducer(
@Value("${spring.kafka.client-id}") String clientId,
ApplicationProperties properties,
- Producer<String, Message> kafkaProducer,
+ Producer<String, String> kafkaProducer,
ConfigurableApplicationContext applicationContext)
{
return
private final String id;
private final String topic;
private final Duration throttle;
- private final Producer<String, Message> producer;
+ private final Producer<String, String> producer;
private final Thread workerThread;
private final Runnable closeCallback;
String id,
String topic,
Duration throttle,
- Producer<String, Message> producer,
+ Producer<String, String> producer,
Runnable closeCallback)
{
this.id = id;
{
final long time = System.currentTimeMillis();
- final ProducerRecord<String, Message> record = new ProducerRecord<>(
+ final ProducerRecord<String, String> record = new ProducerRecord<>(
topic, // Topic
key, // Key
- value // Value
+ value.toString() // Value
);
producer.send(record, (metadata, e) ->
buffer-memory: 33554432
batch-size: 16384
compression-type: gzip
- value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
properties:
delivery-timeout: 10s
max-block: 5s