From 195b2fb77b9ca8d8dc23956817c910b15d1c8a70 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 9 Jan 2023 19:10:17 -0600 Subject: [PATCH] Add sample app for Pulsar Functions See #135 #136 --- ...eckClasspathForProhibitedDependencies.java | 1 + settings.gradle | 2 + .../src/main/asciidoc/pulsar-function.adoc | 22 ++ .../sample-pulsar-functions/.gitignore | 1 + .../sample-pulsar-functions/README.adoc | 227 ++++++++++++++++++ .../sample-pulsar-functions/cassandra-init.sh | 10 + .../docker-compose.yml | 53 ++++ .../download-connectors.sh | 7 + .../sample-signup-app/build.gradle | 37 +++ .../sample/signup/SignupApplication.java | 87 +++++++ .../sample/signup/config/AppConfig.java | 94 ++++++++ .../pulsar/sample/signup/model/Customer.java | 24 ++ .../pulsar/sample/signup/model/Signup.java | 20 ++ .../sample/signup/model/SignupGenerator.java | 40 +++ .../sample/signup/model/SignupTier.java | 33 +++ .../src/main/resources/application.yml | 5 + .../sample-signup-function/build.gradle | 31 +++ .../SignupFunction.java | 62 +++++ .../model/Customer.java | 41 ++++ .../model/Signup.java | 33 +++ .../model/SignupTier.java | 23 ++ 21 files changed, 853 insertions(+) create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/.gitignore create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/README.adoc create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/cassandra-init.sh create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/docker-compose.yml create mode 100755 spring-pulsar-sample-apps/sample-pulsar-functions/download-connectors.sh create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/build.gradle create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/SignupApplication.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/Customer.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/Signup.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/SignupGenerator.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/SignupTier.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/resources/application.yml create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/build.gradle create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/SignupFunction.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/Customer.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/Signup.java create mode 100644 spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/SignupTier.java diff --git a/buildSrc/src/main/java/org/springframework/pulsar/gradle/classpath/LenientCheckClasspathForProhibitedDependencies.java b/buildSrc/src/main/java/org/springframework/pulsar/gradle/classpath/LenientCheckClasspathForProhibitedDependencies.java index 9be26042..eece241f 100644 --- a/buildSrc/src/main/java/org/springframework/pulsar/gradle/classpath/LenientCheckClasspathForProhibitedDependencies.java +++ b/buildSrc/src/main/java/org/springframework/pulsar/gradle/classpath/LenientCheckClasspathForProhibitedDependencies.java @@ -34,6 +34,7 @@ public class LenientCheckClasspathForProhibitedDependencies extends CheckClasspa private static Set OVERRIDE_PROHIBITED_DEPENDENCIES = Set.of( "javax.validation:validation-api", "javax.ws.rs:javax.ws.rs-api", + "javax.inject:javax.inject", "javax.xml.bind:jaxb-api", "commons-logging:commons-logging"); diff --git a/settings.gradle b/settings.gradle index e941a353..81acd562 100644 --- a/settings.gradle +++ b/settings.gradle @@ -31,6 +31,8 @@ include 'spring-pulsar-spring-boot-starter' include 'spring-pulsar-reactive-spring-boot-starter' include 'spring-pulsar-sample-apps:sample-app1' include 'spring-pulsar-sample-apps:sample-app2' +include 'spring-pulsar-sample-apps:sample-pulsar-functions:sample-signup-app' +include 'spring-pulsar-sample-apps:sample-pulsar-functions:sample-signup-function' include 'spring-pulsar-sample-apps:sample-reactive' include 'spring-pulsar-docs' include 'spring-pulsar-spring-cloud-stream-binder' diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc index 16564365..cc88f6ac 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc @@ -69,6 +69,28 @@ The action that occurs during the create/update operation is dependent on path " === Built-in Source and Sinks Apache Pulsar provides many source and sink connectors out-of-the-box, aka built-in connectors. To use a built-in connector simply set the `archive` to `builtin://` (eg `builtin://rabbit`). +== Custom functions +The details on how to develop and package custom functions can be found in the https://pulsar.apache.org/docs/2.10.x/functions-develop[Pulsar docs]. +However, at a high-level, the requirements are as follows: + +* Code uses Java8 +* Code implements either `java.util.Function` or `org.apache.pulsar.functions.api.Function` +* Packaged as uber jar + +Once the function is built and packaged, there are several ways to make it available for function registration. + +=== file:// +The jar file can be uploaded to the server and then referenced via `file://` in the `jar` property of the function config + +=== local +The jar file can remain local and then referenced via the local path in the `jar` property of the function config. + +=== http:// +The jar file can be made available via HTTP server and then referenced via `http(s)://` in the `jar` property of the function config + +=== function:// +The jar file can be uploaded to the Pulsar package manager and then referenced via `function://` in the `jar` property of the function config + == Examples Here are some examples that show how to configure a `PulsarSource` bean which results in the `PulsarFunctionAdministration` auto-creating the backing Pulsar source connector. diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/.gitignore b/spring-pulsar-sample-apps/sample-pulsar-functions/.gitignore new file mode 100644 index 00000000..5cf335a6 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/.gitignore @@ -0,0 +1 @@ +connectors/ diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/README.adoc b/spring-pulsar-sample-apps/sample-pulsar-functions/README.adoc new file mode 100644 index 00000000..96f90fff --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/README.adoc @@ -0,0 +1,227 @@ += Sample Stream Pipeline with Pulsar Functions + +:curdir: {docdir}/guides/pulsar-functions + +A sample appfootnote:[Inspired by the sample app in https://streamnative.io/blog/release/2022-09-21-announcing-spring-for-apache-pulsar/] that leverages the built-in Pulsar Functions support in Spring for Apache Pulsar to create a streaming pipeline to handle user signups. + +The app consists of a Rabbit source, a Pulsar function, and a Cassandra sink with the following details: + +* Messages sent to RabbitMQ `user_signup` queue are sourced into Pulsar `user-signup` topic +* Messages from `user-signup` topic are fed into Pulsar Signup function which: +** the signup tier count is incremented and logged +** `ENTERPRISE` tier signups result in a customer message in `customer-onboard` Pulsar topic +* Messages from `customer-onboard` topic are sinked into `customer_onboard` Cassandra table + +== Pre-requisites +**** +* Ability to run Docker containers locally and Docker Compose installed - see https://docs.docker.com/compose/install/[here] for more details +**** + +== Steps + +**** +NOTE: All commands are expected to be executed from the directory this document lives in [.small]#link:{curdir}[]# +**** + +=== Build app and function +Build the application and function libs with the following command: +---- +../../gradlew clean build +---- + +=== Download connectors +The connector libs are not included in the Docker image, download them one time by executing the following script: + +[source,shell] +---- +./download-connectors.sh +---- + +=== Start services +Start the Pulsar, RabbitMQ, and Cassandra services using Docker Compose with the following command: +---- +docker-compose up -d +---- + +.Verify `rabbitmq` is ready by executing the following command +---- +docker logs rabbitmq | grep "Server startup complete" +---- +.which should produce +---- +2023-01-04 21:06:39.692935+00:00 [info] <0.721.0> Server startup complete; 4 plugins started. +---- + +.Verify `pulsar` is ready by executing the following command +---- +docker logs pulsar 2>&1 | grep "messaging service is ready" +---- +.which should produce +---- +2023-01-04T21:39:57 [main] INFO org.apache.pulsar.broker.PulsarService - messaging service is ready, bootstrap_seconds=4 +2023-01-04T21:39:57 [main] INFO org.apache.pulsar.broker.PulsarService - messaging service is ready, bootstrap service port = 8080, broker url= pulsar://localhost:6650, cluster=standalone, configs=org.apache.pulsar.broker.... +---- + +.Verify `pulsar` has the connectors installed by executing the following command +---- +curl -s http://localhost:8080/admin/v2/functions/connectors +---- +.which should produce +[source,json] +---- +[ + { + "name": "cassandra", + "description": "Writes data into Cassandra", + "sinkClass": "org.apache.pulsar.io.cassandra.CassandraStringSink", + "sinkConfigClass": "org.apache.pulsar.io.cassandra.CassandraSinkConfig" + }, + { + "name": "rabbitmq", + "description": "RabbitMQ source and sink connector", + "sourceClass": "org.apache.pulsar.io.rabbitmq.RabbitMQSource", + "sinkClass": "org.apache.pulsar.io.rabbitmq.RabbitMQSink", + "sourceConfigClass": "org.apache.pulsar.io.rabbitmq.RabbitMQSourceConfig", + "sinkConfigClass": "org.apache.pulsar.io.rabbitmq.RabbitMQSinkConfig" + } +] +---- + +.Verify `cassandra` is ready by executing the following command +---- +docker logs cassandra 2>&1 | grep "Creating replication strategy sample_pulsar_functions_keyspace params KeyspaceParams" +---- +.which should produce +---- +INFO [Native-Transport-Requests-5] 2023-01-04 21:40:48,635 Keyspace.java:381 - Creating replication strategy sample_pulsar_functions_keyspace params KeyspaceParams{durable_writes=true, replication=ReplicationParams{class=org.apache.cassandra.locator.SimpleStrategy, replication_factor=1}} +---- +NOTE: It will typically take about 60 seconds for `cassandra` to complete its startup process so if the above fails, try again after 60 seconds. + +At this point the following services are up and running: + +* `cassandra` +** **keyspace:** `sample_pulsar_functions_keyspace` +** **table:** `customer_onboard` +* `rabbitmq` +** management UI +* `pulsar` standalone +** function support enabled +** `cassandra` and `rabbit` connectors installed + +=== Start application +The sample app registers the Pulsar Functions which effectively create the streaming pipeline. + +.Run the sample app within your IDE or by executing the following command +---- +cd sample-signup-app && ../../../gradlew bootRun +---- +.which should produce +---- +[main] PulsarFunctionAdministration : Creating 'UserSignupFunction' function (using local archive: /Users/cbono/repos/spring-pulsar/spring-pulsar-sample-apps/sample-pulsar-functions/signup-function/target/signup-function-0.0.1-SNAPSHOT.jar) +[main] PulsarFunctionAdministration : Creating 'CustomerOnboardCassandraSink' sink (using local archive: builtin://cassandra) +[main] PulsarFunctionAdministration : Creating 'UserSignupRabbitSource' source (using local archive: builtin://rabbitmq) +[main] SignupApplication : Started SignupApplication in 6.485 seconds (process running for 6.839) +---- + +.Verify the functions are actually registered by executing the following Pulsar commands +---- +docker exec -it pulsar bin/pulsar-admin sources list +docker exec -it pulsar bin/pulsar-admin functions list +docker exec -it pulsar bin/pulsar-admin sinks list +---- +.which should produce +---- +[ + "UserSignupRabbitSource" +] +UserSignupFunction +[ + "CustomerOnboardCassandraSink" +] +---- + +=== Verify pipeline +The app produces a random user signup record to the RabbitMQ `user_signup` queue every 5 seconds. +It also logs all messages on the `user-signup` and `customer-onboard` Pulsar topics as well as the last 5 emails sent to the `customer_onboard` Cassandra table. + +To verify the pipeline is working simply watch the console log as the app runs. +The output should look like similar to the following: +---- +TO RABBIT user_signup => Signup[signupTier=ENTERPRISE, firstName=Samuel, lastName=Weiss, email=samuel.weiss@robutenia.eu, signupTimestamp=1673236049021] +FROM PULSAR user-signup => Signup[signupTier=ENTERPRISE, firstName=Samuel, lastName=Weiss, email=samuel.weiss@robutenia.eu, signupTimestamp=1673236049021] +FROM PULSAR customer-onboard => Customer[firstName=Samuel, lastName=Weiss, email=samuel.weiss@robutenia.eu, signupTimestamp=1673236049021] +FROM CASSANDRA => latest (5/18) emails: carson.maddox@interdemconsulting.biz, aria.burke@interdemassociates.biz, layla.burks@memortech.com, joshua.chandler@furbainc.com, abigail.cooley@quickerinc.com... + +TO RABBIT user_signup => Signup[signupTier=BASIC, firstName=Arianna, lastName=Edwards, email=arianna.edwards@robutenia.eu, signupTimestamp=1673236054031] +FROM PULSAR user-signup => Signup[signupTier=BASIC, firstName=Arianna, lastName=Edwards, email=arianna.edwards@robutenia.eu, signupTimestamp=1673236054031] + +TO RABBIT user_signup => Signup[signupTier=STANDARD, firstName=Kylie, lastName=Raymond, email=kylie.raymond@yrsa.eu, signupTimestamp=1673236059038] +FROM PULSAR user-signup => Signup[signupTier=STANDARD, firstName=Kylie, lastName=Raymond, email=kylie.raymond@yrsa.eu, signupTimestamp=1673236059038] + +TO RABBIT user_signup => Signup[signupTier=ENTERPRISE, firstName=Nolan, lastName=Floyd, email=nolan.floyd@flyhighassociates.eu, signupTimestamp=1673236064045] +FROM PULSAR user-signup => Signup[signupTier=ENTERPRISE, firstName=Nolan, lastName=Floyd, email=nolan.floyd@flyhighassociates.eu, signupTimestamp=1673236064045] +FROM PULSAR customer-onboard => Customer[firstName=Nolan, lastName=Floyd, email=nolan.floyd@flyhighassociates.eu, signupTimestamp=1673236064045] +FROM CASSANDRA => latest (5/19) emails: carson.maddox@interdemconsulting.biz, nolan.floyd@flyhighassociates.eu, aria.burke@interdemassociates.biz, layla.burks@memortech.com, joshua.chandler@furbainc.com... +202 +---- + +.View Pulsar function logs by executing the following command +---- +docker logs pulsar +---- +.which should contain signup logs such as +---- +Processing Signup(signupTier=ENTERPRISE, firstName=Gavin, lastName=Wilson, email=gavin.wilson@beans.eu, signupTimestamp=1673196872351) + ENTERPRISE signup count: 1 +Converting to Signup(signupTier=ENTERPRISE, firstName=Gavin, lastName=Wilson, email=gavin.wilson@beans.eu, signupTimestamp=1673196872351) +Processing Signup(signupTier=FREE, firstName=Nevaeh, lastName=Sexton, email=nevaeh.sexton@linger.eu, signupTimestamp=1673196877357) + FREE signup count: 1 +Processing Signup(signupTier=ENTERPRISE, firstName=Charlotte, lastName=Beach, email=charlotte.beach@quickerconsulting.eu, signupTimestamp=1673196882364) + ENTERPRISE signup count: 2 +Converting to Signup(signupTier=ENTERPRISE, firstName=Charlotte, lastName=Beach, email=charlotte.beach@quickerconsulting.eu, signupTimestamp=1673196882364) +---- + +==== Select from Cassandra +Each `ENTERPRISE` signup should result in a record in the Cassandra table. +To inspect all customer onboard records you can query the Cassandra table. + +.Invoke the `CQLSH` utility on the cassandra container w/ the following command +---- +docker exec -it cassandra cqlsh cassandra +---- +.From the `cqlsh>` prompt execute the following +---- +use sample_pulsar_functions_keyspace; +select * from customer_onboard; +exit; +---- +.which should produce output similar to +---- + customer_email | customer_details +--------------------------------------+----------------------------------------------------------------------------------------------------------------------------- + molly.mckay@morsem.com | {"firstName":"Molly","lastName":"Mckay","email":"molly.mckay@morsem.com","signupTimestamp":1673196862339} + gavin.wilson@beans.eu | {"firstName":"Gavin","lastName":"Wilson","email":"gavin.wilson@beans.eu","signupTimestamp":1673196872351} + ryan.ramsey@felics.biz | {"firstName":"Ryan","lastName":"Ramsey","email":"ryan.ramsey@felics.biz","signupTimestamp":1673196892373} +---- + +=== Stop app and services +Stop the sample app by entering `CTRL-C` in terminal it is running in. + +Stop all running services using Docker Compose with the following command: +---- +docker-compose down -v +---- + +== Useful commands + +.Details about source + docker exec -ti pulsar bin/pulsar-admin sources get --name UserSignupRabbitSource + +.Details about sink + docker exec -ti pulsar bin/pulsar-admin sinks get --name CustomerOnboardCassandraSink + +.Details about function + docker exec -ti pulsar bin/pulsar-admin functions get --name UserSignupFunction + +.Consume messages from output topic of Signup function + docker exec -ti pulsar bin/pulsar-client consume customer_onboard -s "co-sub1" -p "Earliest" -n 100 diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/cassandra-init.sh b/spring-pulsar-sample-apps/sample-pulsar-functions/cassandra-init.sh new file mode 100644 index 00000000..4f1ca081 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/cassandra-init.sh @@ -0,0 +1,10 @@ +CQL="CREATE KEYSPACE IF NOT EXISTS sample_pulsar_functions_keyspace WITH replication = {'class':'SimpleStrategy', 'replication_factor':1}; +USE sample_pulsar_functions_keyspace; +CREATE TABLE customer_onboard (customer_email text PRIMARY KEY, customer_details text);" + +until echo $CQL | cqlsh; do + echo "cqlsh: Cassandra is unavailable to initialize - will retry later" + sleep 20 +done & + +exec /usr/local/bin/docker-entrypoint.sh "$@" diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/docker-compose.yml b/spring-pulsar-sample-apps/sample-pulsar-functions/docker-compose.yml new file mode 100644 index 00000000..45efbe1d --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/docker-compose.yml @@ -0,0 +1,53 @@ +version: '3.5' + +services: + pulsar: + image: kezhenxu94/pulsar:latest + command: bin/pulsar standalone + ports: + - "6650:6650" + - "8080:8080" + container_name: pulsar + hostname: pulsar + environment: + PULSAR_MEM: " -Xms512m -Xmx512m -XX:MaxDirectMemorySize=1g" + volumes: + - ./connectors:/pulsar/connectors + - pulsardata:/pulsar/data + - pulsarconf:/pulsar/conf + restart: on-failure + networks: + - demo_pulsar_pipeline + + rabbitmq: + image: rabbitmq:3.11-management + ports: + - "15672:15672" + - "5672:5672" + container_name: rabbitmq + hostname: rabbitmq + restart: on-failure + networks: + - demo_pulsar_pipeline + + cassandra: + image: cassandra:latest + command: sh /cassandra-init.sh + ports: + - "9042:9042" + container_name: cassandra + hostname: cassandra + volumes: + - ./cassandra-init.sh:/cassandra-init.sh + restart: on-failure + networks: + - demo_pulsar_pipeline + +networks: + demo_pulsar_pipeline: + name: demo_pulsar_pipeline + driver: bridge + +volumes: + pulsardata: + pulsarconf: diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/download-connectors.sh b/spring-pulsar-sample-apps/sample-pulsar-functions/download-connectors.sh new file mode 100755 index 00000000..7aa73a20 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/download-connectors.sh @@ -0,0 +1,7 @@ +#!/bin/bash + +mkdir connectors +cd connectors +wget https://archive.apache.org/dist/pulsar/pulsar-2.10.2/connectors/pulsar-io-cassandra-2.10.2.nar +wget https://archive.apache.org/dist/pulsar/pulsar-2.10.2/connectors/pulsar-io-rabbitmq-2.10.2.nar +cd .. diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/build.gradle b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/build.gradle new file mode 100644 index 00000000..01118da8 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/build.gradle @@ -0,0 +1,37 @@ +plugins { + id 'org.springframework.pulsar.spring-module' + id 'org.springframework.boot' version '3.0.0' +} + +group = 'org.springframework.pulsar.sample' +description = 'Sample Signup App (Pulsar Functions)' + +dependencies { + implementation project(':spring-pulsar-spring-boot-starter') + implementation 'org.springframework.boot:spring-boot-starter-amqp' + implementation 'org.springframework.boot:spring-boot-starter-data-cassandra' + implementation 'com.devskiller:jfairy:0.6.5' + implementation 'com.google.code.findbugs:jsr305' + + // observability + implementation 'org.springframework.boot:spring-boot-starter-actuator' + implementation 'io.micrometer:micrometer-tracing-bridge-brave' + implementation 'io.zipkin.reporter2:zipkin-reporter-brave' + implementation 'io.zipkin.reporter2:zipkin-sender-urlconnection' +} + +bootRun { + jvmArgs = [ + "--add-opens", "java.base/java.lang=ALL-UNNAMED", + "--add-opens", "java.base/java.util=ALL-UNNAMED", + "--add-opens", "java.base/sun.net=ALL-UNNAMED" + ] +} + +project.afterEvaluate { + project.tasks.publishArtifacts.enabled(false) + project.tasks.artifactoryPublish.enabled(false) + project.tasks.publishToOssrh.enabled(false) + project.tasks.publishMavenJavaPublicationToOssrhRepository.enabled(false) + project.tasks.publishAllPublicationsToOssrhRepository.enabled(false) +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/SignupApplication.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/SignupApplication.java new file mode 100644 index 00000000..df9aa131 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/SignupApplication.java @@ -0,0 +1,87 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.sample.signup; + +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.schema.SchemaType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.data.cassandra.core.CassandraTemplate; +import org.springframework.pulsar.annotation.PulsarListener; +import org.springframework.pulsar.sample.signup.model.Customer; +import org.springframework.pulsar.sample.signup.model.Signup; +import org.springframework.pulsar.sample.signup.model.SignupGenerator; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.annotation.Scheduled; + +@SpringBootApplication +@EnableScheduling +public class SignupApplication { + + private final Logger logger = LoggerFactory.getLogger(SignupApplication.class); + + public static void main(String[] args) { + SpringApplication.run(SignupApplication.class, args); + } + + @Autowired + private CassandraTemplate cassandra; + + @Autowired + private RabbitTemplate rabbit; + + @Autowired + private SignupGenerator signupGenerator; + + @Scheduled(initialDelay = 5_000, fixedDelay = 5_000) + void produceSignupToRabbit() { + Signup signup = this.signupGenerator.generate(); + this.rabbit.convertAndSend("user_signup", signup); + this.logger.info("TO RABBIT user_signup => {}", signup); + } + + @PulsarListener(topics = "user-signup", schemaType = SchemaType.JSON, subscriptionName = "pl-us-sub", + subscriptionType = SubscriptionType.Shared) + void logUserSignups(Signup signup) { + this.logger.info("FROM PULSAR user-signup => {}", signup); + } + + @PulsarListener(topics = "customer-onboard", schemaType = SchemaType.JSON, subscriptionName = "pl-co-sub", + subscriptionType = SubscriptionType.Shared) + void logCustomerOnboards(Customer customer) { + this.logger.info("FROM PULSAR customer-onboard => {}", customer); + } + + @Scheduled(initialDelay = 10_000, fixedDelay = 15_000) + void logSinkedOnboardInCassandra() { + List emails = this.cassandra.getCqlOperations() + .queryForList("SELECT customer_email FROM customer_onboard", String.class); + Collections.reverse(emails); + String lastFiveEmails = emails.stream().limit(5).collect(Collectors.joining(", ")); + this.logger.info("FROM CASSANDRA => latest (5/{}) emails: {}...", emails.size(), lastFiveEmails); + } + +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java new file mode 100644 index 00000000..210c03a4 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java @@ -0,0 +1,94 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.sample.signup.config; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.io.SinkConfig; +import org.apache.pulsar.common.io.SourceConfig; + +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.pulsar.function.PulsarFunction; +import org.springframework.pulsar.function.PulsarSink; +import org.springframework.pulsar.function.PulsarSource; +import org.springframework.pulsar.sample.signup.model.SignupGenerator; + +@Configuration(proxyBeanMethods = false) +class AppConfig { + + @Bean + SignupGenerator signupGenerator() { + return new SignupGenerator(); + } + + @Bean + Jackson2JsonMessageConverter jackson2JsonMessageConverter() { + return new Jackson2JsonMessageConverter(); + } + + @Bean + PulsarSource userSignupRabbitSource() { + Map configs = new HashMap<>(); + configs.put("host", "rabbitmq"); + configs.put("port", 5672); + configs.put("virtualHost", "/"); + configs.put("username", "guest"); + configs.put("password", "guest"); + configs.put("queueName", "user_signup"); + configs.put("connectionName", "user_signup_pulsar_source"); + SourceConfig sourceConfig = SourceConfig.builder().tenant("public").namespace("default") + .name("UserSignupRabbitSource").archive("builtin://rabbitmq").topicName("user-signup").configs(configs) + .build(); + return new PulsarSource(sourceConfig, null); + } + + @Bean + PulsarFunction userSignupFunction(@Value("${PWD:.}") String currentRunDir) { + // Abs path differs when run from w/in IDE and on command line (figure it out) + String repoRelativePathToFunctionJar = "/spring-pulsar/spring-pulsar-sample-apps" + + "/sample-pulsar-functions/sample-signup-function" + + "/build/libs/sample-signup-function-0.1.1-SNAPSHOT.jar"; + int idx = currentRunDir.indexOf("/spring-pulsar"); + String absPathToRepo = currentRunDir.substring(0, Math.max(0, idx)); + String absPathToFunctionJar = absPathToRepo + repoRelativePathToFunctionJar; + FunctionConfig functionConfig = FunctionConfig.builder().tenant("public").namespace("default") + .name("UserSignupFunction").className("org.springframework.pulsar.sample.signup.SignupFunction") + .jar(absPathToFunctionJar).inputs(List.of("user-signup")).build(); + return new PulsarFunction(functionConfig, null); + } + + @Bean + PulsarSink customerOnboardCassandraSink() { + Map configs = new HashMap<>(); + configs.put("roots", "cassandra:9042"); + configs.put("keyspace", "sample_pulsar_functions_keyspace"); + configs.put("columnFamily", "customer_onboard"); + configs.put("keyname", "customer_email"); + configs.put("columnName", "customer_details"); + SinkConfig sinkConfig = SinkConfig.builder().tenant("public").namespace("default") + .name("CustomerOnboardCassandraSink").archive("builtin://cassandra").inputs(List.of("customer-onboard")) + .configs(configs).build(); + return new PulsarSink(sinkConfig, null); + } + +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/Customer.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/Customer.java new file mode 100644 index 00000000..9ebe751e --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/Customer.java @@ -0,0 +1,24 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.sample.signup.model; + +public record Customer(String firstName, String lastName, String email, long signupTimestamp) { + + public static Customer from(Signup signup) { + return new Customer(signup.firstName(), signup.lastName(), signup.email(), signup.signupTimestamp()); + } +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/Signup.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/Signup.java new file mode 100644 index 00000000..8b0c6592 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/Signup.java @@ -0,0 +1,20 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.sample.signup.model; + +public record Signup(SignupTier signupTier, String firstName, String lastName, String email, long signupTimestamp) { +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/SignupGenerator.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/SignupGenerator.java new file mode 100644 index 00000000..e095c518 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/SignupGenerator.java @@ -0,0 +1,40 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.sample.signup.model; + +import java.util.Random; + +import com.devskiller.jfairy.Fairy; +import com.devskiller.jfairy.producer.person.Person; + +public class SignupGenerator { + + private static final Random random = new Random(); + + private static final Fairy fairy = Fairy.create(); + + public Signup generate() { + Person person = fairy.person(); + return new Signup(generateSignupTier(), person.getFirstName(), person.getLastName(), person.getCompanyEmail(), + System.currentTimeMillis()); + } + + private static SignupTier generateSignupTier() { + return SignupTier.values()[random.nextInt(SignupTier.values().length)]; + } + +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/SignupTier.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/SignupTier.java new file mode 100644 index 00000000..faafe33a --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/model/SignupTier.java @@ -0,0 +1,33 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.sample.signup.model; + +public enum SignupTier { + + /** Free tier. */ + FREE, + + /** Basic tier. */ + BASIC, + + /** Standard tier. */ + STANDARD, + + /** Enterprise tier. */ + ENTERPRISE + +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/resources/application.yml b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/resources/application.yml new file mode 100644 index 00000000..093986ab --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/resources/application.yml @@ -0,0 +1,5 @@ +spring: + cassandra: + contact-points: localhost + keyspace-name: sample_pulsar_functions_keyspace + local-datacenter: datacenter1 diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/build.gradle b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/build.gradle new file mode 100644 index 00000000..82372f49 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/build.gradle @@ -0,0 +1,31 @@ +plugins { + id 'java' +} + +group = 'org.springframework.pulsar.sample' +description = 'Sample Signup Pulsar Function' +sourceCompatibility = '1.8' + +repositories { + mavenCentral() +} + +dependencies { + implementation 'org.apache.pulsar:pulsar-client-all:2.10.2' + implementation 'org.apache.pulsar:pulsar-functions-api:2.10.2' + compileOnly 'org.projectlombok:lombok:1.18.24' + annotationProcessor 'org.projectlombok:lombok:1.18.24' +} + +// Customization of jar to make a simple Uber function jar +jar { + manifest { + attributes 'Main-Class': 'org.springframework.pulsar.sample.signup.SignupFunction' + } + duplicatesStrategy = DuplicatesStrategy.EXCLUDE + from sourceSets.main.output + dependsOn configurations.runtimeClasspath + from(configurations.runtimeClasspath.findAll { it.name.endsWith('jar') }.collect { zipTree(it) }) { + exclude 'META-INF/*.RSA', 'META-INF/*.SF', 'META-INF/*.DSA' + } +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/SignupFunction.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/SignupFunction.java new file mode 100644 index 00000000..9f6cbdc3 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/SignupFunction.java @@ -0,0 +1,62 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.pulsar.sample.signup; + +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.functions.api.Context; +import org.apache.pulsar.functions.api.Function; + +import org.springframework.pulsar.sample.signup.model.Customer; +import org.springframework.pulsar.sample.signup.model.Signup; +import org.springframework.pulsar.sample.signup.model.SignupTier; + +public class SignupFunction implements Function { + + @Override + public Void process(Signup signup, Context context) { + log("Processing " + signup); + + // Count and log the signups for tier + SignupTier tier = signup.getSignupTier(); + context.incrCounter(tier.name(), 1L); + long count = context.getCounter(tier.name()); + log(String.format(" %s signup count: %d", tier.name(), count)); + + // Create customer onboard for enterprise signups + if (tier == SignupTier.ENTERPRISE) { + Customer customer = Customer.from(signup); + log("Converting to " + signup); + try { + context.newOutputMessage("customer-onboard", Schema.JSON(Customer.class)) + .key(customer.getEmail()) + .value(customer) + .eventTime(System.currentTimeMillis()) + .send(); + } catch (PulsarClientException e) { + e.printStackTrace(); + throw new RuntimeException(e.getMessage(), e); + } + } + return null; + } + + private void log(String msg) { + // Typically logging is done via context.getLogger() but that logs to a topic + // which is not helpful during dev/debugging - sending to console for that reason + System.out.println(msg); + } +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/Customer.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/Customer.java new file mode 100644 index 00000000..eecf3899 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/Customer.java @@ -0,0 +1,41 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.sample.signup.model; + +import lombok.AllArgsConstructor; +import lombok.Data; + +@Data +@AllArgsConstructor +public class Customer { + + private String firstName; + + private String lastName; + + private String email; + + private long signupTimestamp; + + public static Customer from(Signup signup) { + return new Customer( + signup.getFirstName(), + signup.getLastName(), + signup.getEmail(), + signup.getSignupTimestamp()); + } +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/Signup.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/Signup.java new file mode 100644 index 00000000..ad5c1118 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/Signup.java @@ -0,0 +1,33 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.sample.signup.model; + +import lombok.Data; + +@Data +public class Signup { + + private SignupTier signupTier; + + private String firstName; + + private String lastName; + + private String email; + + private long signupTimestamp; +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/SignupTier.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/SignupTier.java new file mode 100644 index 00000000..46dbb7d3 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/src/main/java/org.springframework.pulsar.sample.signup/model/SignupTier.java @@ -0,0 +1,23 @@ +/* + * Copyright 2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.pulsar.sample.signup.model; + +public enum SignupTier { + FREE, + BASIC, + STANDARD, + ENTERPRISE +}