diff --git a/multi-binder-samples/kafka-multi-binder-jaas/README.adoc b/multi-binder-samples/kafka-multi-binder-jaas/README.adoc index d73a05e..b578713 100644 --- a/multi-binder-samples/kafka-multi-binder-jaas/README.adoc +++ b/multi-binder-samples/kafka-multi-binder-jaas/README.adoc @@ -97,15 +97,15 @@ $ .//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 @@ $ .//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. \ No newline at end of file +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. \ No newline at end of file diff --git a/multi-binder-samples/kafka-multi-binder-jaas/pom.xml b/multi-binder-samples/kafka-multi-binder-jaas/pom.xml index 9cad661..6a4833b 100644 --- a/multi-binder-samples/kafka-multi-binder-jaas/pom.xml +++ b/multi-binder-samples/kafka-multi-binder-jaas/pom.xml @@ -3,7 +3,7 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 - kafka-multi-binder-jaas + kafka-multi-binder-jaas-two-clusters 0.0.1-SNAPSHOT jar diff --git a/multi-binder-samples/kafka-multi-binder-jaas/src/main/java/multibinder/kafka/jaas/MultiBinderKafkaJaasSample.java b/multi-binder-samples/kafka-multi-binder-jaas/src/main/java/multibinder/kafka/jaas/MultiBinderKafkaJaasSample.java index 4decbe8..6f1a226 100644 --- a/multi-binder-samples/kafka-multi-binder-jaas/src/main/java/multibinder/kafka/jaas/MultiBinderKafkaJaasSample.java +++ b/multi-binder-samples/kafka-multi-binder-jaas/src/main/java/multibinder/kafka/jaas/MultiBinderKafkaJaasSample.java @@ -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 receive() { - return foo -> foo; + return String::toUpperCase; } @Bean public Function receive1() { return foo -> foo; } + + @Bean + public Supplier supply() { + return () -> "foo-" + random.nextInt(); + } + + @Bean + public Consumer consume() { + return System.out::println; + } } } \ No newline at end of file diff --git a/multi-binder-samples/kafka-multi-binder-jaas/src/main/resources/application.yml b/multi-binder-samples/kafka-multi-binder-jaas/src/main/resources/application.yml index 4b0294a..02ea0a1 100644 --- a/multi-binder-samples/kafka-multi-binder-jaas/src/main/resources/application.yml +++ b/multi-binder-samples/kafka-multi-binder-jaas/src/main/resources/application.yml @@ -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