Chris Bono
@@ -34,6 +34,7 @@ public class LenientCheckClasspathForProhibitedDependencies extends CheckClasspa
|
||||
private static Set<String> 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");
|
||||
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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://<connector-type>` (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.
|
||||
|
||||
1
spring-pulsar-sample-apps/sample-pulsar-functions/.gitignore
vendored
Normal file
1
spring-pulsar-sample-apps/sample-pulsar-functions/.gitignore
vendored
Normal file
@@ -0,0 +1 @@
|
||||
connectors/
|
||||
227
spring-pulsar-sample-apps/sample-pulsar-functions/README.adoc
Normal file
227
spring-pulsar-sample-apps/sample-pulsar-functions/README.adoc
Normal file
@@ -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
|
||||
@@ -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 "$@"
|
||||
@@ -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:
|
||||
7
spring-pulsar-sample-apps/sample-pulsar-functions/download-connectors.sh
Executable file
7
spring-pulsar-sample-apps/sample-pulsar-functions/download-connectors.sh
Executable file
@@ -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 ..
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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<String> 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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, Object> 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<String, Object> 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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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) {
|
||||
}
|
||||
@@ -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)];
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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
|
||||
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
spring:
|
||||
cassandra:
|
||||
contact-points: localhost
|
||||
keyspace-name: sample_pulsar_functions_keyspace
|
||||
local-datacenter: datacenter1
|
||||
@@ -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'
|
||||
}
|
||||
}
|
||||
@@ -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<Signup, Void> {
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user