SpringCloud/spring-cloud-stream-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update kafka-multi-binder-jaas

Browse Source 
Update kafka-multi-binder-jaas to demonstrate how to use two Kafka clusters
with different jaas configs within a single application
This commit is contained in:
Soby Chacko
2021-07-07 15:16:32 -04:00
parent 2447b7c167
commit 8d11589518
4 changed files with 67 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
multi-binder-samples/kafka-multi-binder-jaas/README.adoc
View File

@@ -97,15 +97,15 @@ $ ./<PATH-TO-CLUSTER-1>/bin/kafka-server-start.sh config/server.properties
```
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin-secret"
user_admin="admin-secret";
username="foo"
password="foo-secret"
user_foo="foo-secret";
};
Client {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin-secret";
username="foo"
password="foo-secret";
};
```
@@ -114,9 +114,9 @@ Client {
```
Server {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin-secret"
user_admin="admin-secret";
username="foo"
password="foo-secret"
user_foo="foo-secret";
};
```
@@ -181,8 +181,6 @@ $ ./<PATH-TO-CLUSTER-2>/bin/kafka-server-start.sh config/server.properties
## Running the application
The application's configuration is matched with the plaintext config that is set above (Please review the yml file)
The application contains two `StreamListener` methods. The first one receives records from a topic in cluster-1 and output that to a topic in cluster-2.
The second `StreamListener` method receives records from a topic in cluster-2 and output that to a topic in cluster-1.
@@ -190,7 +188,17 @@ Cluster-1 input topic is named as kafka1-in and output topic is kafka1-out. Simi
Run the application `MultiBinderKafkaJaasSample` on the IDE or from CLI.
## Verify that the application is running
## Running and verifying the application
Review the application's configuration for kafka client security settings.
The application contains a supplier (`supply`) that produces data to a topic in Kafka-1.
From this topic, a function ('receive1) will consume records and then produce to a topic in Kafka-2.
We provide another consumer ('consume`) in the application that consumes data from this topic in Kafka-2 and then log on the console.
This is the function definition: `supply;receive;consume`
You can simply run the main application and then see the test consumer prints data on the console.
* Terminal 5
@@ -204,13 +212,17 @@ KafkaClient {
};
```
You can also change the function definition to this `spring.cloud.function.definition: receive`.
In this case, we only run a single function, and it is your responsibility to send data to the listening topic.
For that, you can follow the instructions below.
* Terminal 6 - Produce data to kafka1 - input.
$ `export KAFKA_OPTS="-Djava.security.auth.login.config=/home/username/kafka_client_jaas.conf"`
Go to the kafka installation directory (It doesn't matter if you are in cluster-1 or cluster-2 directories for the instructions below)
$ `./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka1-in --producer.config=config/producer.properties`
$ `./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic foo --producer.config=config/producer.properties`
* Terminal 7 (or split the above terminal into 2 window panes) - Consume data from kafka2 - output where the above data is expected to come through the processor.
@@ -218,28 +230,9 @@ $ `export KAFKA_OPTS="-Djava.security.auth.login.config=/home/username/kafka_cli
Go to the kafka installation directory (It doesn't matter if you are in cluster-1 or cluster-2 at for the instructions below)
$ `./bin/kafka-console-consumer.sh --topic kafka2-out --consumer.config=config/consumer.properties --bootstrap-server=localhost:9093`
$ `./bin/kafka-console-consumer.sh --topic bar --consumer.config=config/consumer.properties --bootstrap-server=localhost:9093`
* Now start adding some text to the terminal session where you are running console producer on kafka1 input topic (`foo`).
Then verify that, you see the same exact text, but in uppercase, on the terminal session where you are running the console consumer on kafka2 output topic (`bar`).
* Terminal 8 - Produce data to kafka2 - input.
$ `export KAFKA_OPTS="-Djava.security.auth.login.config=/home/username/kafka_client_jaas.conf"`
Go to the kafka installation directory (It doesn't matter if you are in cluster-1 or cluster-2 at for the instructions below)
$ `./bin/kafka-console-producer.sh --broker-list localhost:9093 --topic kafka2-in --producer.config=config/producer.properties`
* Terminal 9 (or split the above terminal into 2 window panes) - Consume data from kafka1 - output where the above data is expected to come through the second processor in the application.
$ `export KAFKA_OPTS="-Djava.security.auth.login.config=/home/username/kafka_client_jaas.conf"`
Go to the kafka installation directory (It doesn't matter if you are in cluster-1 or cluster-2 at for the instructions below)
$ `./bin/kafka-console-consumer.sh --topic kafka1-out --consumer.config=config/consumer.properties --bootstrap-server=localhost:9092`
* Now start adding some text to the terminal session where you are running console producer on kafka1-in.
Then verify that, you see the same exact text on the the terminal session where you are running the console consumer on kafka2-out.
Similarly, start adding some text to the terminal session where you are running console producer on kafka2-in.
Then verify that, you see the same exact text on the the terminal session where you are running the console consumer on kafka1-out.
PS: Once you are done with the testing, remember to stop the application, console consumers and producers and your local kafka clusters used for testing.
PS: Once you are done with the testing, remember to stop the application, console consumer and producer (if you are running them), and your local kafka clusters used for testing.

2
multi-binder-samples/kafka-multi-binder-jaas/pom.xml
View File

@@ -3,7 +3,7 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>kafka-multi-binder-jaas</artifactId>
<artifactId>kafka-multi-binder-jaas-two-clusters</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>

17
multi-binder-samples/kafka-multi-binder-jaas/src/main/java/multibinder/kafka/jaas/MultiBinderKafkaJaasSample.java
View File

@@ -16,7 +16,10 @@
package multibinder.kafka.jaas;
import java.util.Random;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -31,15 +34,27 @@ public class MultiBinderKafkaJaasSample {
static class Foo {
Random random = new Random();
@Bean
public Function<String, String> receive() {
return foo -> foo;
return String::toUpperCase;
}
@Bean
public Function<String, String> receive1() {
return foo -> foo;
}
@Bean
public Supplier<String> supply() {
return () -> "foo-" + random.nextInt();
}
@Bean
public Consumer<String> consume() {
return System.out::println;
}
}
}

38
multi-binder-samples/kafka-multi-binder-jaas/src/main/resources/application.yml
View File

@@ -1,18 +1,18 @@
spring.cloud.stream:
function.definition: receive;receive1
function.definition: supply;receive;consume
bindings:
supply-out-0:
destination: foo
binder: kafka1
receive-in-0:
destination: kafka1-in
destination: foo
binder: kafka1
receive-out-0:
destination: kafka2-out
destination: bar
binder: kafka2
receive1-in-0:
destination: kafka2-in
consume-in-0:
destination: bar
binder: kafka2
receive1-out-0:
destination: kafka1-out
binder: kafka1
binders:
kafka1:
type: kafka
@@ -21,10 +21,14 @@ spring.cloud.stream:
cloud:
stream:
kafka:
binder.brokers: localhost:9092
binder.jaas.loginModule: org.apache.kafka.common.security.plain.PlainLoginModule
binder.jaas.options.username: admin
binder.jaas.options.password: admin-secret
binder:
brokers: localhost:9092
configuration.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\";"
# The following properties should not be used if you have separate authorization contexts for the two clusters.
# Only use them if they are identical (in which case, you don't need the above property - sasl.jaas.config.
# binder.jaas.loginModule: org.apache.kafka.common.security.plain.PlainLoginModule
# binder.jaas.options.username: admin
# binder.jaas.options.password: admin-secret
kafka2:
type: kafka
environment:
@@ -33,11 +37,13 @@ spring.cloud.stream:
stream:
kafka:
binder:
zkNodes: localhost:2182
brokers: localhost:9093
jaas.loginModule: org.apache.kafka.common.security.plain.PlainLoginModule
jaas.options.username: admin
jaas.options.password: admin-secret
configuration.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"foo\" password=\"foo-secret\";"
# The following properties should not be used if you have separate authorization contexts for the two clusters.
# Only use them if they are identical (in which case, you don't need the above property - sasl.jaas.config.
# jaas.loginModule: org.apache.kafka.common.security.plain.PlainLoginModule
# jaas.options.username: foo
# jaas.options.password: foo-secret
kafka.binder:
configuration:
security.protocol: SASL_PLAINTEXT