Spring Cloud Stream

https://github.com/spring-cloud/spring-cloud-stream
Admin
Spring Cloud Stream is a framework for building message-driven microservice applications that...
Tokens:482,837
Snippets:3,042
Trust Score:9.5
License:Apache-2.0
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### Application startup output with RabbitMQ

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/preface.adoc

Expected console output when starting the application with RabbitMQ. Shows queue provisioning, connection setup, and successful startup.

```text
	--- [ main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg, bound to: input
	--- [ main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [localhost:5672]
	--- [ main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#2a3a299:0/SimpleConnection@66c83fc8. . .
	. . .
	--- [ main] o.s.i.a.i.AmqpInboundChannelAdapter      : started inbound.input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg
	. . .
	--- [ main] c.e.l.LoggingConsumerApplication         : Started LoggingConsumerApplication in 2.531 seconds (JVM running for 2.897)
```

--------------------------------

### Start Kafka cluster, Control Center UI, and Schema Registry

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/tools/kafka/docker-compose/README.adoc

Starts the cluster, UI, and schema registry by chaining all three Docker Compose files together.

```shell
docker-compose -f kafka-cluster.yml -f ./control-center-ui.yml -f ./schema-registry.yml up
```

--------------------------------

### Start Kafka cluster and Control Center UI

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/tools/kafka/docker-compose/README.adoc

Starts the brokers and the Confluent Control Center UI. The scripts must be chained together as the UI depends on the brokers.

```shell
docker-compose -f ./kafka-cluster.yml -f ./control-center-ui.yml up
```

--------------------------------

### Start Kafka cluster and Schema Registry

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/tools/kafka/docker-compose/README.adoc

Starts the brokers and the Confluent Schema Registry. The scripts must be chained together as the schema registry depends on the brokers.

```shell
docker-compose -f ./kafka-cluster.yml -f ./schema-registry.yml up
```

--------------------------------

### META-INF/spring.binders file example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/overview-binder-api.adoc

Shows the format of the META-INF/spring.binders file, mapping a binder name to its configuration class.

```text
kafka:\
org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration
```

--------------------------------

### Start Kafka cluster with Zookeeper

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/tools/kafka/docker-compose/README.adoc

Starts a multi-node Kafka cluster with 3 brokers available at localhost:9091,9092,9093 using Zookeeper for leader election. Add -d to run in the foreground.

```shell
docker-compose -f ./kafka-cluster.yml up
```

--------------------------------

### Implement ProvisioningProvider for file messaging

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/overview-custom-binder-impl.adoc

Example ProvisioningProvider implementation that trims the destination name. It returns a FileMessageDestination for both producer and consumer destinations, and throws UnsupportedOperationException for partitioning.

```java
public class FileMessageBinderProvisioner implements ProvisioningProvider<ConsumerProperties, ProducerProperties> {

    @Override
    public ProducerDestination provisionProducerDestination(
            final String name,
            final ProducerProperties properties) {

        return new FileMessageDestination(name);
    }

    @Override
    public ConsumerDestination provisionConsumerDestination(
            final String name,
            final String group,
            final ConsumerProperties properties) {

        return new FileMessageDestination(name);
    }

    private class FileMessageDestination implements ProducerDestination, ConsumerDestination {

        private final String destination;

        private FileMessageDestination(final String destination) {
            this.destination = destination;
        }

        @Override
        public String getName() {
            return destination.trim();
        }

        @Override
        public String getNameForPartition(int partition) {
            throw new UnsupportedOperationException("Partitioning is not implemented for file messaging.");
        }

    }

}
```

--------------------------------

### Register global state store using KafkaStreamsInfrastructureCustomizer

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc

Use this configurer to register a global state store. The example uses KafkaStreamsInfrastructureCustomizer to add a global store to the StreamsBuilder. Avoid calling StreamsBuilderFactoryBean#getObject() to get the StreamsBuilder as it may cause circular dependency issues.

```java
@Bean
public StreamsBuilderFactoryBeanConfigurer customizer() {
    return streamsBuilderFactoryBean -> {
        try {
            streamsBuilderFactoryBean.setInfrastructureCustomizer(new KafkaStreamsInfrastructureCustomizer() {
                  @Override
                  public void configureBuilder(StreamsBuilder builder) {
                      builder.addGlobalStore(
                              ...
                      );
                  }
              });
        }
        catch (Exception e) {

        }
    };
}
```

--------------------------------

### Manually start processor via actuator REST endpoint

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc

Curl command to POST a state change to `STARTED` for the `process3-in-0` binding. Requires Spring Boot actuator dependency on the classpath.

```bash
curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/process3-in-0

```

--------------------------------

### Get host info and query local or remote store

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc

Use getHostInfo to determine which instance hosts a key, then query locally or remotely based on the current host.

```java
org.apache.kafka.streams.state.HostInfo hostInfo = interactiveQueryService.getHostInfo("store-name",
						key, keySerializer);

if (interactiveQueryService.getCurrentHostInfo().equals(hostInfo)) {

    //query from the store that is locally available
}
else {
    //query from the remote host
}
```

--------------------------------

### Kafka Streams binder configuration properties example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/configuration-options.adoc

Example of setting Kafka Streams binder-level configuration properties using the `spring.cloud.stream.kafka.streams.binder.configuration` prefix. These properties apply to the entire application and all processors.

```properties
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000
```

--------------------------------

### Equivalent JAAS file for KafkaClient

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_overview.adoc

JAAS configuration file equivalent to the Spring Boot properties example above. Uses Krb5LoginModule with required flag and options for keytab and principal.

```plaintext
KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka_client.keytab"
    principal="kafka-client-1@EXAMPLE.COM";
};
```

--------------------------------

### Run SimpleConsumerApplication jar with destination property

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc

Run the SimpleConsumerApplication jar to start the Kafka Streams consumer. Use the property spring.cloud.stream.bindings.process-in-0.destination=my-topic to set the destination for the single inbound binding named process-in-0.

```bash
java -jar kstream-consumer-app.jar --spring.cloud.stream.bindings.process-in-0.destination=my-topic

```

--------------------------------

### Customize StreamsBuilderFactoryBean with state listener

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc

Use this configurer to customize the StreamsBuilderFactoryBean before it is started. The example sets a state listener that is invoked on state changes.

```java
@Bean
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
    return sfb -> sfb.setStateListener((newState, oldState) -> {
         //Do some action here!
    });
}
```

--------------------------------

### GET /{subject}/{format}/{version}

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc

Retrieves an existing schema by subject, format, and version.

```APIDOC
## GET /{subject}/{format}/{version}

### Description
Retrieves an existing schema by subject, format, and version.

### Method
GET

### Endpoint
/{subject}/{format}/{version}

### Parameters
#### Path Parameters
- **subject** (string) - Required - The schema subject.
- **format** (string) - Required - The schema format.
- **version** (string) - Required - The schema version.

### Response
#### Success Response (200)
- **id** (string) - The schema ID.
- **subject** (string) - The schema subject.
- **format** (string) - The schema format.
- **version** (string) - The schema version.
- **definition** (string) - The schema definition.

#### Response Example
{
  "id": "string",
  "subject": "string",
  "format": "string",
  "version": "string",
  "definition": "string"
}
```

--------------------------------

### Checkstyle suppressions XML example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/README.adoc

Example suppressions file that suppresses specific rules for specific files. Place it at ${project.root}/src/checkstyle/checkstyle-suppressions.xml.

```xml
<?xml version="1.0"?>
<!DOCTYPE suppressions PUBLIC
		"-//Puppy Crawl//DTD Suppressions 1.1//EN"
		"https://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
<suppressions>
	<suppress files=".*ConfigServerApplication\.java" checks="HideUtilityClassConstructor"/>
	<suppress files=".*ConfigClientWatch\.java" checks="LineLengthCheck"/>
</suppressions>
```

--------------------------------

### Start consumer-in-0 binding

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc

Start the consumer-in-0 binding again. When multiple bindings exist on a single function, invoking these operations on any of them works because they share the same StreamsBuilderFactoryBean.

```bash
curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/consumer-in-0
```

--------------------------------

### Download and run schema registry server

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc

Downloads and runs the standalone schema registry server version 4.0.3. The server is started as a standalone application using the downloaded JAR.

```bash
wget https://repo1.maven.org/maven2/org/springframework/cloud/spring-cloud-stream-schema-registry-server/4.0.3/spring-cloud-stream-schema-registry-server-4.0.3.jar
java -jar ./spring-cloud-stream-schema-registry-server-4.0.3.jar
```

--------------------------------

### Start Kafka Streams processors manually with StreamsBuilderFactoryManager

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors.adoc

Use this bean to start all Kafka Streams processors after disabling auto-startup via spring.kafka.streams.auto-startup=false. The ApplicationRunner calls sbfm.start() to start the processors separately from the main application startup.

```java
@Bean
public ApplicationRunner runner(StreamsBuilderFactoryManager sbfm) {
    return args -> {
        sbfm.start();
    };
}
```

--------------------------------

### Control binding lifecycle with BindingsLifecycleController

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/binding_visualization_control.adoc

Retrieve BindingsLifecycleController from the Spring application context and use it to query and change the state of a binding. The example shows querying the state of 'echo-in-0', asserting it is running, stopping it, and asserting it is not running. It also mentions alternative methods stop, start, pause, and resume.

```java
BindingsLifecycleController bindingsController = context.getBean(BindingsLifecycleController.class);
Binding binding = bindingsController.queryState("echo-in-0");
assertThat(binding.isRunning()).isTrue();
bindingsController.changeState("echo-in-0", State.STOPPED);
//Alternative way of changing state. For convenience we expose start/stop and pause/resume operations.
//bindingsController.stop("echo-in-0")
assertThat(binding.isRunning()).isFalse();
```

--------------------------------

### Get binding details via actuator bindings endpoint

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc

Query the bindings endpoint to see details about all three bindings. The output shows the binding names, groups, and states.

```bash
 curl http://localhost:8080/actuator/bindings | jq .
[
  {
    "bindingName": "consumer-in-0",
    "name": "consumer-in-0",
    "group": "consumer-applicationId",
    "pausable": false,
    "state": "running",
    "paused": false,
    "input": true,
    "extendedInfo": {}
  },
  {
    "bindingName": "function-in-0",
    "name": "function-in-0",
    "group": "function-applicationId",
    "pausable": false,
    "state": "running",
    "paused": false,
    "input": true,
    "extendedInfo": {}
  },
  {
    "bindingName": "function-out-0",
    "name": "function-out-0",
    "group": "function-applicationId",
    "pausable": false,
    "state": "running",
    "paused": false,
    "input": false,
    "extendedInfo": {}
  }
]
```

--------------------------------

### Launch Spring Cloud Stream application with SASL and Kerberos using Spring Boot properties

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_overview.adoc

Command-line example that sets Kafka binder properties for SASL_PLAINTEXT and Kerberos JAAS options. Note that autoCreateTopics is disabled; ensure topics exist or are created administratively. Do not mix with JAAS configuration files.

```bash
 java --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \
   --spring.cloud.stream.bindings.input.destination=stream.ticktock \
   --spring.cloud.stream.kafka.binder.autoCreateTopics=false \
   --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT \
   --spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true \
   --spring.cloud.stream.kafka.binder.jaas.options.storeKey=true \
   --spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab \
   --spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM
```

--------------------------------

### Re-route DLQ messages for partitioned destinations with republishToDlq=false

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc

Spring Boot application for partitioned destinations where RabbitMQ publishes to DLX/DLQ with an x-death header. The example is incomplete in the source, showing only the class definition and constants.

```java
@SpringBootApplication
public class ReRouteDlqApplication {

	private static final String ORIGINAL_QUEUE = "so8400in.so8400";

	private static final String DLQ = ORIGINAL_QUEUE + ".dlq";

	private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot";

	private static final String X_DEATH_HEADER = "x-death";
```

--------------------------------

### Manually start processor using BindingsEndpoint API

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc

Java code that autowires `BindingsEndpoint` and uses an `ApplicationRunner` to call `changeState` with `State.STARTED` for the `process3-in-0` binding. Requires Spring Boot actuator dependency.

```java
@Autowired
BindingsEndpoint endpoint;

@Bean
public ApplicationRunner runner() {
    return args -> {
        endpoint.changeState("process3-in-0", State.STARTED);
    };
}

```

--------------------------------

### Required configuration for transactional Kafka example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_tips.adoc

Required configuration for the transactional Kafka example. Sets transaction prefix, required acks, consumer group and destinations, and disables binder retries by setting maxAttempts to 1.

```yaml
spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix: tx-
spring.cloud.stream.kafka.binder.required-acks=all
spring.cloud.stream.bindings.input-in-0.group=foo
spring.cloud.stream.bindings.input-in-0.destination=input
spring.cloud.stream.bindings.left.destination=left
spring.cloud.stream.bindings.right.destination=right

spring.cloud.stream.kafka.bindings.input-in-0.consumer.maxAttempts=1

```

--------------------------------

### GET /subject/format

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc

Retrieves existing schemas by subject and format.

```APIDOC
## GET /subject/format

### Description
Retrieves existing schemas by subject and format.

### Method
GET

### Endpoint
/subject/format

### Parameters
#### Path Parameters
- **subject** (string) - Required - The schema subject.
- **format** (string) - Required - The schema format.

### Response
#### Success Response (200)
A list of schema objects, each with the following fields:
- **id** (string) - The schema ID.
- **subject** (string) - The schema subject.
- **format** (string) - The schema format.
- **version** (string) - The schema version.
- **definition** (string) - The schema definition.

#### Response Example
[
  {
    "id": "string",
    "subject": "string",
    "format": "string",
    "version": "string",
    "definition": "string"
  }
]
```

--------------------------------

### ReRouteDlqApplication with republishToDlq=true headers

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc

This example shows a Spring Boot application that listens to a DLQ and republishes failed messages up to 3 times using the original exchange and routing key headers added by the republishing recoverer when republishToDlq=true. After exceeding retries, it sends the message to a parking lot queue. The main method runs the application and waits for user input to exit.

```java
@SpringBootApplication
public class ReRouteDlqApplication {

	private static final String ORIGINAL_QUEUE = "so8400in.so8400";

	private static final String DLQ = ORIGINAL_QUEUE + ".dlq";

	private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot";

	private static final String X_ORIGINAL_EXCHANGE_HEADER = RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE;

	private static final String X_ORIGINAL_ROUTING_KEY_HEADER = RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY;

	public static void main(String[] args) throws Exception {
		ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args);
		System.out.println("Press enter to exit");
		System.in.read();
		context.close();
	}

	@Autowired
	private RabbitTemplate rabbitTemplate;

	@RabbitListener(queues = DLQ)
	public void rePublish(Message failedMessage) {
		Map<String, Object> headers = failedMessage.getMessageProperties().getHeaders();
		long retries = failedMessage.getMessageProperties().getRetryCount();
		if (retries < 3) {
			failedMessage.getMessageProperties().incrementRetryCount();
			String exchange = (String) headers.get(X_ORIGINAL_EXCHANGE_HEADER);
			String originalRoutingKey = (String) headers.get(X_ORIGINAL_ROUTING_KEY_HEADER);
			this.rabbitTemplate.send(exchange, originalRoutingKey, failedMessage);
		}
		else {
			this.rabbitTemplate.send(PARKING_LOT, failedMessage);
		}
	}

	@Bean
	public Queue parkingLot() {
		return new Queue(PARKING_LOT);
	}

}
```

--------------------------------

### RabbitMQ dynamic destination configuration callback

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/event-routing.adoc

Example of a NewDestinationBindingCallback bean for the RabbitMQ binder, configuring required groups, queue name group only, auto-bind DLQ, and dead letter queue name.

```java
@Bean
public NewDestinationBindingCallback<RabbitProducerProperties> dynamicConfigurer() {
    return (name, channel, props, extended) -> {
        props.setRequiredGroups("bindThisQueue");
        extended.setQueueNameGroupOnly(true);
        extended.setAutoBindDlq(true);
        extended.setDeadLetterQueueName("myDLQ");
    };
}

```

--------------------------------

### ReRouteDlqApplication with x-death header republishing

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/rabbit/rabbit_dlq.adoc

This example shows a Spring Boot application that listens to a DLQ and republishes failed messages up to 3 times using the x-death header to extract the original exchange and routing key. After exceeding retries, it sends the message to a parking lot queue. The main method runs the application and waits for user input to exit.

```java
public static void main(String[] args) throws Exception {
		ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args);
		System.out.println("Press enter to exit");
		System.in.read();
		context.close();
	}

	@Autowired
	private RabbitTemplate rabbitTemplate;

	@SuppressWarnings("unchecked")
	@RabbitListener(queues = DLQ)
	public void rePublish(Message failedMessage) {
		Map<String, Object> headers = failedMessage.getMessageProperties().getHeaders();
		long retries = failedMessage.getMessageProperties().getRetryCount();
		if (retries < 3) {
			failedMessage.getMessageProperties().incrementRetryCount();
			List<Map<String, ?>> xDeath = (List<Map<String, ?>>) headers.get(X_DEATH_HEADER);
			String exchange = (String) xDeath.get(0).get("exchange");
			List<String> routingKeys = (List<String>) xDeath.get(0).get("routing-keys");
			this.rabbitTemplate.send(exchange, routingKeys.get(0), failedMessage);
		}
		else {
			this.rabbitTemplate.send(PARKING_LOT, failedMessage);
		}
	}

	@Bean
	public Queue parkingLot() {
		return new Queue(PARKING_LOT);
	}

}
```

--------------------------------

### Define three Kafka Streams processors (process1, process2, process3)

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc

Example of three function beans: a `Function`, a `Consumer`, and a `BiFunction`. The `BiFunction` has two input bindings (`process3-in-0` and `process3-in-1`). This is the context for selectively disabling auto-start on one processor.

```java
@Bean
public Function<KStream<?, ?>, KStream<?, ?>> process1() {

}

@Bean
public Consumer<KStream<?, ?>> process2() {

}

@Bean
public BiFunction<KStream<?, ?>, KTable<?, ?>, KStream<?, ?>> process3() {

}

```

--------------------------------

### Example message handler function

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/mechanics.adoc

A simple Function handler that expects a Person and returns a String. Used to illustrate content-type negotiation.

```java
public Function<Person, String> personFunction {..}
```

--------------------------------

### Define AdminClientConfigCustomizer bean

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-binder/admin-client-config-cust.adoc

Provides a bean that customizes admin client properties. The configure method receives a Properties object, and the example sets the security protocol to SASL_SSL.

```java
@Bean
public AdminClientConfigCustomizer adminClientConfigCustomizer() {
    return props -> {
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
    };
}
```

--------------------------------

### GET /{subject}/{format}/{version}

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc

Retrieves an existing schema by subject, format, and version. The response contains the schema object including its ID, subject, format, version, and definition.

```APIDOC
## GET /{subject}/{format}/{version}

### Description
Retrieves an existing schema by subject, format, and version.

### Method
GET

### Endpoint
/{subject}/{format}/{version}

### Parameters
#### Path Parameters
- **subject** (string) - Required - The schema subject
- **format** (string) - Required - The schema format
- **version** (integer) - Required - The schema version

### Response
#### Success Response (200)
- **id** (integer) - The schema ID
- **subject** (string) - The schema subject
- **format** (string) - The schema format
- **version** (integer) - The schema version
- **definition** (string) - The schema definition

#### Response Example
{
  "id": 1,
  "subject": "foo",
  "format": "avro",
  "version": 1,
  "definition": "{\"type\":\"record\",\"name\":\"Foo\",\"fields\":[]}"
}
```

--------------------------------

### Kafka Streams processor with multiple binder types

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc

Example of a Kafka Streams processor that uses KStream, KTable, and GlobalKTable. In a multi-binder scenario, the binder type must be explicitly provided for each binding.

```java
@Bean
public Function<KStream<Long, Order>,
        Function<KTable<Long, Customer>,
                Function<GlobalKTable<Long, Product>, KStream<Long, EnrichedOrder>>>> enrichOrder() {

    ...
}

```

--------------------------------

### Example DLQ error message with exception stacktrace

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc

Shows the error message routed to the DLQ, including the exception stacktrace and the original payload. The message contains details about the failed message, such as headers and delivery attempts.

```text
. . . .
x-exception-stacktrace:	org.springframework.messaging.MessageHandlingException: nested exception is
      org.springframework.messaging.MessagingException: has an error, failedMessage=GenericMessage [payload=byte[15],
      headers={amqp_receivedDeliveryMode=NON_PERSISTENT, amqp_receivedRoutingKey=input.hello, amqp_deliveryTag=1,
      deliveryAttempt=3, amqp_consumerQueue=input.hello, amqp_redelivered=false, id=a15231e6-3f80-677b-5ad7-d4b1e61e486e,
      amqp_consumerTag=amq.ctag-skBFapilvtZhDsn0k3ZmQg, contentType=application/json, timestamp=1522327846136}]
      at org.spring...integ...han...MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:107)
      at. . . . .
Payload: blah
```

--------------------------------

### Test runner for transactional Kafka example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_tips.adoc

Test runner that sends a failing and a good message to the input topic after waiting for user input. Use to verify the transactional behavior and dead-letter handling.

```java
@Bean
public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
    return args -> {
        System.in.read();
        template.send("input", "Fail".getBytes());
        template.send("input", "Good".getBytes());
    };
}

```

--------------------------------

### Get extension properties for a binding

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/binding_visualization_control.adoc

Shows how to retrieve the extension properties for a binding using BindingsLifecycleController.getExtensionProperties(..). The example retrieves KafkaConsumerProperties for a binding named 'test-input-binding'. Note that explicitly defined bindings do not carry the in-0/out-0 suffix.

```java
KafkaConsumerProperties properties = controller.getExtensionProperties("test-input-binding”);
```

--------------------------------

### Access state store in Processor

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/state-store.adoc

Shows how to retrieve a WindowStore from the processor context during init. Note: this approach does not work for global state stores.

```java
Processor<Object, Product>() {

    WindowStore<Object, String> state;

    @Override
    public void init(ProcessorContext processorContext) {
        state = (WindowStore)processorContext.getStateStore("mystate");
    }
    ...
}
```

--------------------------------

### Set start offset to latest for Kafka Streams bindings

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_tips.adoc

Sets the start offset to 'latest' for both input bindings of the BiConsumer. If only one binding should start from the latest offset, leave the other binding out of the configuration. Once committed offsets exist, these settings are not honored.

```properties
spring.cloud.stream.kafka.streams.bindings.myBiConsumer-in-0.consumer.startOffset: latest
spring.cloud.stream.kafka.streams.bindings.myBiConsumer-in-1.consumer.startOffset: latest
```

--------------------------------

### Use DltAwareProcessor to publish failed records to a DLT

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc

This example shows how to use DltAwareProcessor inside a KStream#process call. The constructor takes a Function (business logic), the DLT topic name, and a DltPublishingContext. When the Function throws an exception, the record is sent to the DLT.

```java
@Bean
public java.util.function.Function<KStream<String, String>, KStream<String, String>> process(DltPublishingContext dltSenderContext) {
    return input -> input
        .process(() -> new DltAwareProcessor<>(record -> {
					throw new RuntimeException("error");
				}, "hello-dlt-1", dltPublishingContext));
}
```

--------------------------------

### Define a BinderCustomizer bean to customize binders by type

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/binder-customizer.adoc

Provides a BinderCustomizer bean that checks the binder type and applies customizations. Use this in multi-binder applications where binders reside in different application contexts. The customizations are applied before the application starts using the binders. When there are multiple binders of the same type, use the binder name to filter customization.

```java
@Bean
public BinderCustomizer binderCustomizer() {
    return (binder, binderName) -> {
        if (binder instanceof KafkaMessageChannelBinder kafkaMessageChannelBinder) {
            kafkaMessageChannelBinder.setRebalanceListener(...);
        }
        else if (binder instanceof KStreamBinder) {
            ...
        }
        else if (binder instanceof RabbitMessageChannelBinder) {
            ...
        }
    };
}
```

--------------------------------

### IntelliJ project structure with conventions

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/README.adoc

Shows the spring-cloud-build-tools structure including IntelliJ-specific files: Intellij_Project_Defaults.xml and Intellij_Spring_Boot_Java_Conventions.xml.

```text
└── src
    ├── checkstyle
    │   └── checkstyle-suppressions.xml <3>
    └── main
        └── resources
            ├── checkstyle-header.txt <2>
            ├── checkstyle.xml <1>
            └── intellij
              ├── Intellij_Project_Defaults.xml <4>
              └── Intellij_Spring_Boot_Java_Conventions.xml <5>
```

--------------------------------

### Sample application.properties for DLQ rerouting

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-binder/dlq.adoc

Configuration for the sample rerouting application. It sets the input group and destination to the DLQ topic, output to the original topic, and a parkingLot destination. It also configures the Kafka binder to reset offsets to earliest and includes the x-retries header.

```properties
spring.cloud.stream.bindings.input.group=so8400replay
spring.cloud.stream.bindings.input.destination=error.so8400out.so8400

spring.cloud.stream.bindings.output.destination=so8400out

spring.cloud.stream.bindings.parkingLot.destination=so8400in.parkingLot

spring.cloud.stream.kafka.binder.configuration.auto.offset.reset=earliest

spring.cloud.stream.kafka.binder.headers=x-retries
```

--------------------------------

### Function bean for binding name example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/binding-properties.adoc

Example of a Spring Cloud Stream function bean that creates two bindings: `uppercase-in-0` for input and `uppercase-out-0` for output.

```java
@Bean
public Function<String, String> uppercase() {
	return v -> v.toUpperCase();
}
```

--------------------------------

### Required dependencies for observability

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/observability.adoc

Add spring-boot-starter-actuator and reactor-core-micrometer to enable observability support.

```xml
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
	<groupId>io.projectreactor</groupId>
	<artifactId>reactor-core-micrometer</artifactId>
</dependency>
```

--------------------------------

### Download .editorconfig and .springformat

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/README.adoc

Bash commands to download the .editorconfig and create an empty .springformat file to apply default formatting rules.

```bash
$ curl https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/.editorconfig -o .editorconfig
$ touch .springformat
```

--------------------------------

### Define new input binding with BindingsLifecycleController

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/binding_visualization_control.adoc

Shows how to define a new input binding using BindingsLifecycleController.createInputBinding(..). It retrieves the DefaultBinderFactory and the binder for 'rabbit', creates BindingProperties with destination and group, sets ConsumerProperties, and calls createInputBinding. The method returns RabbitConsumerProperties.

```java
DefaultBinderFactory binderFactory = context.getBean(DefaultBinderFactory.class);
Object binder = binderFactory.getBinder("rabbit", MessageChannel.class);
		
String inputName = "test-output-binding";
		
BindingsLifecycleController controller = context.getBean(BindingsLifecycleController.class);
		
BindingProperties bindingProperties = new BindingProperties();
bindingProperties.setDestination("myDest");
bindingProperties.setGroup("myGroup"); // must define group for input binding
ConsumerProperties consumerProperties = new ConsumerProperties();
bindingProperties.setConsumer(consumerProperties);	
		
RabbitConsumerProperties extendedConsumerProperties = controller.createInputBinding(inputName, "rabbit", bindingProperties);
```

--------------------------------

### GET /schemas/{id}

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc

Retrieves a schema by its ID.

```APIDOC
## GET /schemas/{id}

### Description
Retrieves a schema by its ID.

### Method
GET

### Endpoint
/schemas/{id}

### Parameters
#### Path Parameters
- **id** (string) - Required - The schema ID.

### Response
#### Success Response (200)
- **id** (string) - The schema ID.
- **subject** (string) - The schema subject.
- **format** (string) - The schema format.
- **version** (string) - The schema version.
- **definition** (string) - The schema definition.

#### Response Example
{
  "id": "string",
  "subject": "string",
  "format": "string",
  "version": "string",
  "definition": "string"
}
```

--------------------------------

### Configure application.server property

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc

Set the application.server configuration property to enable host information lookup for interactive queries in multi-instance deployments.

```properties
spring.cloud.stream.kafka.streams.binder.configuration.application.server: <server>:<port>
```

--------------------------------

### Basic binder configuration for mixed functions

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc

Configuration properties that define the function definition and bindings for both the regular process and the Kafka Streams kstreamProcess.

```properties
spring.cloud.function.definition=process;kstreamProcess
spring.cloud.stream.bindings.process-in-0.destination=foo
spring.cloud.stream.bindings.process-out-0.destination=bar
spring.cloud.stream.bindings.kstreamProcess-in-0.destination=bar
spring.cloud.stream.bindings.kstreamProcess-out-0.destination=foobar

```

--------------------------------

### Stop Kafka cluster with Zookeeper

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/tools/kafka/docker-compose/README.adoc

Stops the multi-node Kafka cluster started with the kafka-cluster.yml file.

```shell
docker-compose -f ./kafka-cluster.yml down
```

--------------------------------

### Get KeyQueryMetadata method signature

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc

Method signature for retrieving KeyQueryMetadata for a given store and key.

```java
public <K> KeyQueryMetadata getKeyQueryMetadata(String store, K key, Serializer<K> serializer)
```

--------------------------------

### Get KafkaStreams method signature

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc

Method signature for retrieving the KafkaStreams object associated with a store and key.

```java
public <K> KafkaStreams getKafkaStreams(String store, K key, Serializer<K> serializer)
```

--------------------------------

### JSON message payload for Person

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/preface.adoc

Example JSON representation of the Person class to send as a message to the input destination.

```json
{"name":"Sam Spade"}
```

--------------------------------

### Configure publisher confirms with useConfirmHeader

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/rabbit/rabbit_overview/publisher-confirms.adoc

Properties configuration for enabling publisher confirms using the correlation header. Requires setting spring.rabbitmq.publisher-confirm-type=correlated and spring.rabbitmq.publisher-returns=true. Also enables batch mode, error channel, and DLQ.

```properties
spring.cloud.stream.bindings.input-in-0.group=someGroup
spring.cloud.stream.bindings.input-in-0.consumer.batch-mode=true

spring.cloud.stream.source=output
spring.cloud.stream.bindings.output-out-0.producer.error-channel-enabled=true

spring.cloud.stream.rabbit.bindings.output-out-0.producer.useConfirmHeader=true
spring.cloud.stream.rabbit.bindings.input-in-0.consumer.auto-bind-dlq=true
spring.cloud.stream.rabbit.bindings.input-in-0.consumer.batch-size=10

spring.rabbitmq.publisher-confirm-type=correlated
spring.rabbitmq.publisher-returns=true
```

--------------------------------

### Processor function that returns KStream

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc

The processor function that processes input and returns a KStream. This is the same processor used in other examples.

```java
@Bean
public Function<KStream<Object, String>, KStream<?, WordCount>> process() {

    ...
}
```

--------------------------------

### Configure output binding with partitionKeyExpression and partitionCount

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/overview-partitioning.adoc

Configures an output binding to partition data by setting partitionKeyExpression (a SpEL expression evaluated against the outbound message, here extracting the 'id' header) and partitionCount. This is a valid and typical configuration for partitioned output.

```properties
spring.cloud.stream.bindings.func-out-0.producer.partitionKeyExpression=headers.id
spring.cloud.stream.bindings.func-out-0.producer.partitionCount=5
```

--------------------------------

### Define a Consumer for odd numbers

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/event-routing.adoc

Defines a Consumer<Integer> bean that prints odd values. This is used as a routing target for the MessageRoutingCallback example.

```java
@Bean
public Consumer<Integer> oddHandler() {
    return value -> System.out.println("ODD: " + value);
}

```

--------------------------------

### Configure topic pattern for a KStream binding

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc

Configure the 'process-in-0' binding to consume from topics matching the regular expression pattern 'input.*'.

```properties
spring.cloud.stream.bindings.process-in-0.destination=input.*
```

--------------------------------

### Customize KafkaStreams with uncaught exception handler

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc

Use this configurer to customize the underlying KafkaStreams object. The example sets an uncaught exception handler.

```java
@Bean
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
    return factoryBean -> {
        factoryBean.setKafkaStreamsCustomizer(new KafkaStreamsCustomizer() {
            @Override
            public void customize(KafkaStreams kafkaStreams) {
                kafkaStreams.setUncaughtExceptionHandler((t, e) -> {

                });
            }
        });
    };
}
```

--------------------------------

### Configure input binding for partitioning

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/overview-partitioning.adoc

Configures an input binding to receive partitioned data. Sets the consumer 'partitioned' property to true and specifies the instance index and count. The instanceIndex must be unique and between 0 and instanceCount-1.

```properties
spring.cloud.stream.bindings.uppercase-in-0.consumer.partitioned=true
spring.cloud.stream.instanceIndex=3
spring.cloud.stream.instanceCount=5
```

--------------------------------

### GET /schemas/{id}

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc

Retrieves an existing schema by its ID. The response contains the schema object including its ID, subject, format, version, and definition.

```APIDOC
## GET /schemas/{id}

### Description
Retrieves an existing schema by its ID.

### Method
GET

### Endpoint
/schemas/{id}

### Parameters
#### Path Parameters
- **id** (integer) - Required - The schema ID

### Response
#### Success Response (200)
- **id** (integer) - The schema ID
- **subject** (string) - The schema subject
- **format** (string) - The schema format
- **version** (integer) - The schema version
- **definition** (string) - The schema definition

#### Response Example
{
  "id": 1,
  "subject": "foo",
  "format": "avro",
  "version": 1,
  "definition": "{\"type\":\"record\",\"name\":\"Foo\",\"fields\":[]}"
}
```

--------------------------------

### Disable auto-startup for process3 via consumer binding (process3-in-1)

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc

Alternative property to set `auto-startup` to `false` on the `process3-in-1` consumer binding. Either binding is sufficient; setting both is optional for clarity.

```properties
spring.cloud.stream.bindings.process3-in-1.consumer.auto-startup: false

```

--------------------------------

### GET /{subject}/{format}

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc

Retrieves an existing schema by subject and format. The response contains the schema object including its ID, subject, format, version, and definition.

```APIDOC
## GET /{subject}/{format}

### Description
Retrieves an existing schema by subject and format.

### Method
GET

### Endpoint
/{subject}/{format}

### Parameters
#### Path Parameters
- **subject** (string) - Required - The schema subject
- **format** (string) - Required - The schema format

### Response
#### Success Response (200)
- **id** (integer) - The schema ID
- **subject** (string) - The schema subject
- **format** (string) - The schema format
- **version** (integer) - The schema version
- **definition** (string) - The schema definition

#### Response Example
{
  "id": 1,
  "subject": "foo",
  "format": "avro",
  "version": 1,
  "definition": "{\"type\":\"record\",\"name\":\"Foo\",\"fields\":[]}"
}
```

--------------------------------

### Per-binding binder configuration

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/multiple-binders.adoc

Configures individual bindings to use specific binders. Here, the input binding uses Kafka and the output binding uses RabbitMQ, overriding the default binder.

```properties
spring.cloud.stream.bindings.input.binder=kafka
spring.cloud.stream.bindings.output.binder=rabbit
```

--------------------------------

### Curried function composition example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc

A curried Function bean (curriedFoo) and a regular Function bean (bar) that can be composed. The curried form is allowed for the first function in the definition.

```java
@Bean
public Function<KStream<String, String>, Function<KTable<String, String>, KStream<String, String>>> curriedFoo() {
    return a -> b ->
            a.join(b, (value1, value2) -> value1 + value2);
}

@Bean
public Function<KStream<String, String>, KStream<String, String>> bar() {
    return input -> input.mapValues(value -> value + "From-anotherFooFunc");
}
```

--------------------------------

### Activate a profile for a binder environment

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/multiple-systems.adoc

Shows how to activate a specific Spring profile for a particular binder environment using the spring.profiles.active property.

```yaml
environment:
    spring:
        profiles:
           active: myBinderProfile

```

--------------------------------

### POST message to StreamBridge REST endpoint

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/producing-and-consuming-messages.adoc

Curl command to POST a plain text message to the REST endpoint of the WebSourceApplication example, demonstrating sending data to a dynamic destination.

```bash
curl -H "Content-Type: text/plain" -X POST -d "hello from the other side" http://localhost:8080/
```

--------------------------------

### Implement FileMessageBinder by extending AbstractMessageChannelBinder

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/spring-cloud-stream/overview-custom-binder-impl.adoc

Extend AbstractMessageChannelBinder to implement a custom binder. Override createProducerMessageHandler to write messages to a file and createConsumerEndpoint to return a FileMessageProducer. The generic argument specifies the ProvisioningProvider.

```java
public class FileMessageBinder extends AbstractMessageChannelBinder<ConsumerProperties, ProducerProperties, FileMessageBinderProvisioner> {

    public FileMessageBinder(
            String[] headersToEmbed,
            FileMessageBinderProvisioner provisioningProvider) {

        super(headersToEmbed, provisioningProvider);
    }

    @Override
    protected MessageHandler createProducerMessageHandler(
            final ProducerDestination destination,
            final ProducerProperties producerProperties,
            final MessageChannel errorChannel) throws Exception {

        return message -> {
            String fileName = destination.getName();
            String payload = new String((byte[])message.getPayload()) + "\n";

            try {
                Files.write(Paths.get(fileName), payload.getBytes(), CREATE, APPEND);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        };
    }

    @Override
    protected MessageProducer createConsumerEndpoint(
            final ConsumerDestination destination,
            final String group,
            final ConsumerProperties properties) throws Exception {

        return new FileMessageProducer(destination);
    }

}
```

--------------------------------

### Kafka partition consumer application main method and listener bean

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-binder/partitions.adoc

Defines the main method that runs the Spring Boot application without a web server, and a listener bean that consumes messages from a Kafka topic and prints the received partition.

```java
public static void main(String[] args) {
    new SpringApplicationBuilder(KafkaPartitionConsumerApplication.class)
        .web(WebApplicationType.NONE)
        .run(args);
}

@Bean
public Consumer<Message<String>> listen() {
    return message -> {
        int partition = (int) message.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION);
        System.out.println(message + " received from partition " + partition);
    };
}

}
```

--------------------------------

### Consumer function with String payload for DLQ serializer example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_tips.adoc

A consumer function that processes String payloads. Used to illustrate the need for a custom serializer when the payload type is not byte[].

```java
@Bean
public Consumer<String> processData() {
  return s -> {
     throw new RuntimeException();
  };
}
```

--------------------------------

### Run wordcount processor with Kafka Streams binder

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc

Run the wordcount processor application, consuming from the Kafka topic 'words' and publishing results to 'counts'. The binding names 'process-in-0' and 'process-out-0' are the defaults for a Function bean named 'process'.

```bash
java -jar wordcount-processor.jar --spring.cloud.stream.bindings.process-in-0.destination=words --spring.cloud.stream.bindings.process-out-0.destination=counts
```

--------------------------------

### Consumer function that throws exception for DLQ example

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_tips.adoc

A trivial consumer function that throws a RuntimeException for every record. Used to demonstrate the DLQ feature in Spring Cloud Stream.

```java
@Bean
public Consumer<byte[]> processData() {
  return s -> {
     throw new RuntimeException();
  };
}
```

--------------------------------

### POST /

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc

Registers a new schema with the Schema Registry Server. The request body must include subject, format, and definition. The response returns the created schema object with its assigned ID and version.

```APIDOC
## POST /

### Description
Registers a new schema with the Schema Registry Server.

### Method
POST

### Endpoint
/

### Parameters
#### Request Body
- **subject** (string) - Required - The schema subject
- **format** (string) - Required - The schema format
- **definition** (string) - Required - The schema definition

### Response
#### Success Response (200)
- **id** (integer) - The schema ID
- **subject** (string) - The schema subject
- **format** (string) - The schema format
- **version** (integer) - The schema version
- **definition** (string) - The schema definition

#### Response Example
{
  "id": 1,
  "subject": "foo",
  "format": "avro",
  "version": 1,
  "definition": "{\"type\":\"record\",\"name\":\"Foo\",\"fields\":[]}"
}
```

--------------------------------

### Define DeadLetterPublishingRecoverer bean

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_tips.adoc

Defines a DeadLetterPublishingRecoverer bean that uses a KafkaOperations template to send messages to the dead letter topic. This is the third bean in the deserialization error handling setup.

```java
	@Bean
	public DeadLetterPublishingRecoverer publisher(KafkaOperations bytesTemplate) {
		return new DeadLetterPublishingRecoverer(bytesTemplate);
	}
```

--------------------------------

### Consume data from Kafka topic

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/tools/kafka/docker-compose/README.adoc

Runs the Kafka console consumer to consume data from the input-topic. Data entered on the producer console should be echoed here.

```shell
docker exec -it broker1 /bin/kafka-console-consumer --bootstrap-server localhost:29091 --topic input-topic
```

--------------------------------

### Define a String consumer function

Source: https://github.com/spring-cloud/spring-cloud-stream/blob/main/docs/modules/ROOT/pages/kafka/kafka_tips.adoc

Defines a trivial consumer function that takes a String parameter. Used as the starting point for the recipe; the function itself is not the focus but the context for configuring deserialization error handling.

```java
@Bean
public Consumer<String> functionName() {
    return s -> {
        System.out.println(s);
    };
}
```
Tokens:
Raw