In this blog, we will experiment with extracting data from SAP Integration Suite API and using AWS Kinesis Data Stream for data ingestion and then have an insight into the collected data with Apache Flink. Also, reacting to anomaly detection of the data.

SAP Integration Suite API provides different sets of APIs such as Integration Content which has enough functions to build a CI/CD pipeline for interface development, Partner Directory provides a flexible role base logical, Security artifacts, and Message Processing Log... We use a few OData endpoints in the Message Processing Log and Integration Content in this blog, consuming API sets is the same except the required role for each API set is different – the role is to authorize the OAuth client to consume APIs. You should consult SAP doc for more detail on the required roles of each API set.

We have 2 parts. SAP Integration Suite (CPI) is the focus of this part where we implement data collecting and reacting - undeploy the artifact which causes permanent errors that we get notification from AWS. In the second part, we will implement an AWS data pipeline to ingress and analyze data and then send out a notification (AWS SNS) if there is something that exceeds the predefined threshold. With collected data, we can also send it to a data lake to study (ML) and can make a better decision in SAP Message Monitor or SAP Security artifact rotation.

The collecting workflow

In a period, we call Message Processing Log OData API to get all messages and for each error message we will call another endpoint to enrich the message with more detail of the error, and then streaming messages to AWS Kinesis Data Stream by using AWS adapter.

Let’s say we have a 2-minute timer, and we have 5000 messages which include 1000 error messages (just an example), in the ideal case we must make 1 + 1000 API calls in a 2-minute window. What if it could not make it? We may increase the timer to have a bigger window to collect data, but it won’t be lucky all the time.

Like any long-run workflow, it’s better to decouple it into asynchronous parts. We can use any event/message service to leverage the decoupling, there are many options in the market, and it depends on each company. We are using AWS and prefer to use Event Bus of Amazon EventBridge as it natively integrates with many target services in AWS. As I know so far there are 3 AWS adapters available on SAP Integration Suite but only the Advantco AWS adapter supports EventBridge now.

In the download package, we also provide an implementation of Event Mesh for your reference, but you will see why we choose a cloud native event service over Event Mesh. With Event Mesh, it simply does the decoupling that why at Integration flow #2 we consume both Success and Error messages from the broker to route them to AWS Kinesis. Amazon EventBridge Event Bus is also an event router where we can route the Success messages to Kinesis inside AWS (we will have details in part 2 of the blog), so we only need to consume the Error messages in Integration flow #2 to enrich with detailed error and route it to the final destination AWS Kinesis.

Another reason and the most important why we chose the Advantco AWS adapter for the implementation is because it’s the only adapter that supports the IAM role – the security best practice in AWS.

We have 2 integration flows:

• Integration flow 1: Collect all messages and split them one by one then send it out to a SQS topic.
• Integration flow 2: Asynchronously receive messages from the SQS topic, enrich the message with the error detail, and then send to a Kinesis Data Stream. Multiple threads are counted here, which helps to handle the high throughput.

Some screenshots of the integration flows are below. You can download the whole package of integration artifacts [here](https://github.com/quachtd/sap_is_aws_data_integration/blob/main/AWS Data Integration 1.0.1.zip) for more details of the implementation. It’s just a proof of concept, you have to handle error gracefully and add more logic if needed.

Integration flow 1 – MessageProcessing_CI2AWS_OB

Integration flow 2 – MessageProcessing_CI2AWS_GetMoreDetail_OB
As I mentioned above with EventBridge Event Bus we route Success messages directly to Kinesis in AWS, for the Error messages we route them to AWS SQS that is why we use AWS SQS as the sender adapter here.
We are supposed to make the integration flow simpler with OData adapter, but the OData receiver adapter doesn’t support navigation of the API which is why we use HTTP adapter and require an additional Integration Process for the enrichment step.

The reacting workflow

Any exceed of the predefined threshold will trigger an SNS message in AWS, and then the subscribed SQS topic will receive the message (configured in AWS). In SAP Integration Suite, we consume the message and consider it as a notification for remediation.

In this implementation, if any integration flow keeps repeating errors for 4 hours continuously then a notification is triggered (in AWS which we will cover in part2) and the integration flow will be undeployed and a notification message will be sent out to inform the support team.

We have 1 integration flow to accomplish it:

• Integration flow 3: Receive a notification message from an SQS topic, then undeploy the integration flow where the integration id is from the message payload, and finally send out an SNS message to inform the support team. Email/Mobile notification from the SNS topic is configured in AWS, not a part of this integration flow.

The screenshot of the integration flow is below.

Integration flow 3 – IntegrationContent_AWS2CI_UndeployRuntimeArtifact_IB

That’s it for the first part. We will continue with the second part of working on AWS.
All you need to try out this implementation:

• [Download the package of artifacts.](https://github.com/quachtd/sap_is_aws_data_integration/blob/main/AWS Data Integration 1.0.1.zip)
• Look at the API sets and configure the access as I mentioned above.
• Reach out to Advantco for the trial version of AWS adapter and they will also help you to understand the configuration of AWS adapter (EventBridge, SQS, SNS, Kinesis).

Hope you find something new from this blog and enjoy it.

The config is different for Neo and Cloud Foundry (CF) environment. This post is for OAuth2 with authorization type of client id and secret. We can use OAuth2 with other grant type or certificate, the config is similar.

Cloud Integration, Neo

#1 Register API Client on cockpit
Cockpit > Security > OAuth > then Clients tab > create a new Client.

looking for Subscription with "tmn"node, and select Grant of Client Credentials. As the screenshot.

#2 Assign specific roles (required by API endpoints) to the API client.
Looking for roles that required by the API endpoint. In this case, I use MessageProcessLogs and the roles for Neo are: IntegrationOperationServer.read, NodeManager.read

Cockpit > Security > OAuth > Authorizations

Hit button Show Assignments for oauth_client_<created client id> to assign role. Looking for Application "tmn" and assign roles to it.

#3 Test it with Postman
We have client id and client secret. Now we need to get token endpoint for OAuth, go to cockpit > Security > OAuth > go to Branding tab> there is a token endpoint here.

From SAP Cloud Integration API doc, we have the endpoint such as /MessageLoggingLogs but we don't know the full URL. Basically the full url is: the url where you access CPI tenant + "/api" + the endpoint.
Or you can find it from cockpit > Applications > Subscriptions > click on the applicaiton with "tmn" > you will see the url with /api.

Cloud Integration, Cloud Foundry

#1 Create a Instance for API Client
Looking for roles that required by the API endpoint. In this case, I use MessageProcessLogs and the role for CF is: MonitorDataRead.

Note: we can reuse the existing Instance, but we have to maintain required roles for different API endpoints. So, add MonitorDataRead to the existing one.

Go to Cockpit > Services > Instances and Subscriptions > create a new instance of "Process Integration Runtime" with "api" plan, and then select role MonitorDataRead (this is important, don't miss it), and "Client Credentials" as grant type.

#2 Create a Service Key from the created instance.
Choose "ClientId/Secret" as the Key Type.

#3 Test with Postman:

From the previous step, we have everything from client id, secret, url (hostname) of the endpoint, and token endpoint for Postman.
Using Oauth with Postman is similar to the part of testing with Neo.

References

• https://help.sap.com/docs/cloud-integration/sap-cloud-integration/message-processing-logs
• https://help.sap.com/docs/cloud-integration/sap-cloud-integration/setting-up-oauth-inbound-authentication-with-client-credentials-grant-for-api-clients
• https://help.sap.com/docs/cloud-integration/sap-cloud-integration/setting-up-inbound-http-connections-for-api-clients

1.       Generate CA and key for ELK stack

./bin/elasticsearch-certutil ca

2.       Generate certs and its key for nodes (in the cluster, do the same for each node)

./bin/elasticsearch-certutil cert --ca elastic-stack-ca.p12

3.       Config the node inter communication - elasticsearch.yml

xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate xpack.security.transport.ssl.client_authentication: required
xpack.security.transport.ssl.keystore.path: elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: elastic-certificates.p12

if there is a password for the private key then add it to elasticsearch keystore.

./bin/elasticsearch-keystore add xpack.security.transport.ssl.keystore.secure_password

./bin/elasticsearch-keystore add xpack.security.transport.ssl.truststore.secure_password

4.       Generate cert and its key for https config

./bin/elasticsearch-certutil http

It generated elasticsearch-ssl-http.zip file:

/elasticsearch
|_ README.txt
|_ http.p12
|_ sample-elasticsearch.yml

/kibana
|_ README.txt
|_ elasticsearch-ca.pem
|_ sample-kibana.yml

5.       From Elasticsearch, use the key http.p12 from previous step - elasticsearch.yml

xpack.security.http.ssl.enabled: true
xpack.security.http.ssl.keystore.path: http.p12

if there is a password for the private key, then add it to elasticsearch keystore.

./bin/elasticsearch-keystore add xpack.security.http.ssl.keystore.secure_password

6.       From Kibana, use the cert elasticsearch-ca.pem from the previous step – kibana.yml

elasticsearch.ssl.certificateAuthorities: elasticsearch-ca.pem

elasticsearch.hosts: https://:9200

References:
https://www.elastic.co/guide/en/elasticsearch/reference/current/security-basic-setup.html

https://www.elastic.co/guide/en/elasticsearch/reference/current/security-basic-setup-https.html

Updated: 7/22/2024 - Symptom: "Error during ping operation:Error while silently connecting: org.w3c.www.protocol.http.HttpException: Peer sent alert: Alert Fatal: handshake failure", SAP note 3424764

At the time of writing this blog – Election week 2020, Confluent Cloud Shema Registry using Let’s Encrypt to sign the certificates for Schema Registry (HTTPS endpoint), it uses TLS 1.2, ECDHE_RSA with P-256, and AES_256_GCM.

And it’s not working with SAP PO 7.5 latest SP 19. There is a list of default cipher suites, but it’s not including the one above. For more information or those steps below not working, refer to this note 2284059.

The solution is to update the cipher suits list. Follow these steps:

1.       Identify the problem. Follow this note 2616423

2.       Make sure the JCE policy files on the server are unlimited. Follow this note 1240081

3.       Add a system property “iaik.security.ssl.configFile” to point custom-defined cipher suites. Follow this note 2708581

4.       Restart Java AS to get the new system property effect.

Implementation Steps:

Note: Don’t forget to import certs chain of the target endpoint to the Netweaver Keystore.

1.       Identify the problem

• Identify the certificate of the request endpoint.

Using Chrome Developer Tool (Ctrl + Shift + I)

Or from website https://www.ssllabs.com/ssltest/index.html

• With an error of handshake in SAP PO trace or XPI Inspector – start XPI Inspector and then trigger a request message to the target endpoint.

SSLException: Peer sent alert: Alert Fatal: handshake failure

Or error of limited JCE policy
Unable to encrypt SSL message: java.security.InvalidKeyException: Illegal key size

2.       Edit/Validate file java.security to the JCE policy unlimited

\usr\sap\<SID>\SYS\exe\jvm\<platform>\sapjvm_8.1.065\sapjvm_8\jre\lib\security

3.       Add property “Diaik.security.ssl.configFile”

Create a file SSLContext.properties – place at anywhere in the server.

#
client.allowLegacyRenegotiation=true
extension=signature_algorithms
extension=server_name.noncritical
extension=elliptic_curves
extension=ec_point_formats
securityProvider=iaik.security.ssl.ECCelerateProvider
#
# enable cipher suites with ECDHE key exchange (unlimited strength)
cipherSuite=TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256
cipherSuite=TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA
cipherSuite=TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256
cipherSuite=TLS_ECDHE_RSA_WITH_CAMELLIA_128_CBC_SHA256
cipherSuite=TLS_ECDHE_RSA_WITH_CAMELLIA_128_GCM_SHA256
cipherSuite=TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384
cipherSuite=TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA
cipherSuite=TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384
cipherSuite=TLS_ECDHE_RSA_WITH_CAMELLIA_256_CBC_SHA384
cipherSuite=TLS_ECDHE_RSA_WITH_CAMELLIA_256_GCM_SHA384
#
# keep default cipher suites as fallback
cipherSuite=TLS_RSA_WITH_AES_128_GCM_SHA256
cipherSuite=TLS_RSA_WITH_AES_128_CBC_SHA
cipherSuite=TLS_RSA_WITH_AES_128_CBC_SHA256
cipherSuite=TLS_RSA_WITH_CAMELLIA_128_GCM_SHA256
cipherSuite=TLS_RSA_WITH_CAMELLIA_128_CBC_SHA
cipherSuite=TLS_RSA_WITH_CAMELLIA_128_CBC_SHA256
cipherSuite=TLS_RSA_WITH_AES_256_GCM_SHA384
cipherSuite=TLS_RSA_WITH_AES_256_CBC_SHA
cipherSuite=TLS_RSA_WITH_AES_256_CBC_SHA256
cipherSuite=TLS_RSA_WITH_CAMELLIA_256_GCM_SHA384
cipherSuite=TLS_RSA_WITH_CAMELLIA_256_CBC_SHA
cipherSuite=TLS_RSA_WITH_CAMELLIA_256_CBC_SHA256
cipherSuite=SSL_RSA_WITH_3DES_EDE_CBC_SHA
cipherSuite=SSL_RSA_WITH_RC4_128_SHA
#
# enable old&slow DHE cipher suites (you don't want them with limited strength 1024-bit DHE at all)
cipherSuite=TLS_DHE_RSA_WITH_AES_128_GCM_SHA256
cipherSuite=TLS_DHE_RSA_WITH_AES_128_CBC_SHA
cipherSuite=TLS_DHE_RSA_WITH_AES_128_CBC_SHA256
cipherSuite=TLS_DHE_RSA_WITH_AES_256_GCM_SHA384
cipherSuite=TLS_DHE_RSA_WITH_AES_256_CBC_SHA
cipherSuite=TLS_DHE_RSA_WITH_AES_256_CBC_SHA256

Add System property – NWA > Configuration > Infrastructure > Java System Properties > Additional VM Parameters

WINDOWS
-Diaik.security.ssl.configFile=file:/d:/tmp/SSLContext.properties

UNIX
-Diaik.security.ssl.configFile=file:/tmp/SSLContext.properties

4.       The restart Java AS.

Validate existing installed drivers

Go to url http://hostname:/nwa/classloader

Request the existing SDA file (from Basis) if there are notice drivers that using with JMS/JDBC channels in the system.

If it is the first time deploy any driver to the system then get com.sap.aii.adapter.lib.sda from the SAPXI3RDPARTY*.SCA (download from software center)

Add driver file to the SDA file

Such as we are going to use JDBC for MySQL 5.1.41 then go to MySQL website to download JDBC driver for that version.

Use SDA Marker Tool to add the Jar file to the existing SDA file - SAP note 1987079

java -jar sdaMakerTool.jar

The tool actually update 2 places in the SDA file:

1. {sda file}\lib - add Jar file to lib folder

2. {sda file}\server - add one line to provider.xml

We can manually edit the SDA file to add Jar file and edit provider.xml if we don't want to use SDA Marker Tool

Many SAP Notes just in cases this not working: 1138877, 850116, 831162, 1987079, 1931899, 2513069

In this blog we will go through steps that using GCP ADC (application default credential) to impersonate another service account to generate the access token.

1. Create a VM

gcloud compute instances create instance-1 --machine-type=f1-micro --zone=us-central1-a --scopes=https://www.googleapis.com/auth/cloud-platform

Validate the Cloud API access scopes is "Allow full access to all Cloud APIs"

Take note the service account from the created VM, we will use it in the next step.

The default service account for the compute is {project number}-compute@developer.gserviceaccount.com

2. Assign role roles/iam.serviceAccountTokenCreator to the service account.

3. Create 2 services accounts for testing purpose

Create the service account gcs-only@goingdelit.iam.gserviceaccount.com

Repeat to create another service account bigquery-only@goingdelit.iam.gserviceaccount.com - but don't grant any role to this user to test the access token

4. Install "gcloud" and connect it to the GCP project in the VM

Follow this guide to install it.

https://cloud.google.com/sdk/install

Don't forgot the last step to connect to GCP via "gcloud init"

5. Get the access token of the ADC

This step have to run inside the VM.

curl "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token" -H "Metadata-Flavor: Google"

The output looks like this

Similar with gcloud command "gcloud auth application-default print-access-token"

6. Enable "IAM Service Account Credentials API"

Open the URL below and enable the API.

https://console.developers.google.com/apis/api/iamcredentials.googleapis.com/overview

7. Generate the access for the service account "gcs-only"

This step we can run anywhere. In our local machine, make a HTTP request to this end point to generate the access token for the service account "gcs-only".

POST https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{email of the service account}:generateAccessToken

curl --location --request POST 'https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/gcs-only@goingdelit.iam.gserviceaccount.com:generateAccessToken' \
--header 'Authorization: Bearer ya29.c.Kn_ZB1Iucwnl1R8u5OJtjMioQE94-_TPv8RaEbzr8AAXG1TeBZT4WY_G9xCDwIFFkoLdYgcOz6AgRN1hpXDP7uQSmhEQocZH9-sJa52DnZXvT39PFdclhNvej1dDxTw9WgiXpI_e74xge1Ss4i2ZfBt2EPxsMjR9jWNVY5ptZdDD' \
--header 'Content-Type: application/json' \
--data-raw '{
  "scope": [
      "https://www.googleapis.com/auth/cloud-platform"
  ]
}'

Note the Bearer Token is from the step #5

Finally we have the access token for the service account "gcs-only", we will use it to perform a storage API test in next step.

8. Test the generated access token

Create a Bucket and upload some files to the bucket.

Make a GET request to this endpoint to list all objects in the bucket

GET https://storage.googleapis.com/storage/v1/b/{bucket name}/o

Repeat step #7 to generate access token for the service account "bigquery-only" - it won't work as the impersonated service account don't have the appropriate role to do operation on Storage Objects (look back step #3).

References

Create short-lived service account credentials

OAuth 2.0 Scopes for Google APIs

ADC - Application default credentials

Generate Access Token API

Access Instance Metadata Endpoint

The Kafka REST Proxy provides REST API to produce and consume messages and view metadata of cluster, broker, partition, and topic.

Using REST API is the fastest way to experiment with producing and consuming messages from Kafka broker. But it is not simple to use this API set, as with the same API endpoint, it works for different serialization formats. In this post, we use Confluent 5.5, which adding support of Protobuf and JSON Schema to the Schema Registry, and we will use API endpoint to produce messages to a Kafka topic with Protobuf schema, and use API endpoints to consume messages from the topic.

Content Types

Content types are the first and essential thing to look at when using the API set. It used for Content-Type and Accept headers of the HTTP request.

This REST API has three versions v1, v2, and v3, but version v3 is in the preview stage, and we do not look at it here

application/vnd.kafka.{Embeded format}.{API version}+json

Here are the samples:

The Avro content type is application/vnd.kafka.avro.v2+json

The Protobuf content type is application/vnd.kafka.protobuf.v2+json

The JSON schema content type is application/vnd.kafka.jsonschema.v2+json

We will see how we use it in the next sections.

Message Schema

First, we define a schema for producer and consumer, we use the same for both here, but it can involve the schema any time without any impact of other ends. That is the purpose of using schema. Here is the schema of Protobuf format that we use in this post.

message value_salesorder_topic {
  required uint32 orderId = 1;
  optional string orderType = 2;
  required uint32 plant = 3;
  optional string soldToCustomer = 4;
  optional string targetSystem = 5;
}

Use the Confluent Control Center to set the schema for the topic

Produce Message

We will send messages to this Kafka topic “salesorder_topic”.

The endpoint

POST /topic/salesorder_topic

Content-Type

We use Protobuf schema and version 2 of the API. And this is the Content-Type header.

application/vnd.kafka.protobuf.v2+json

Before making the HTTP request to send a message, we need the schema ID that we set to the topic from the previous step.

Make a request to the Schema Registry:

curl --location --request GET 'http://{SchemaRegistryHost}:{SchemaRegistryPort}/subjects/salesorder_topic-value/versions/1'

Response: we got the schema ID 41

{
    "subject": "salesorder_topic-value",
    "version": 1,
    "id": 41,
    "schemaType": "PROTOBUF",
    "schema": "\nmessage value_salesorder_topic {\n  required uint32 orderId = 1;\n  optional string orderType = 2;\n  required uint32 plant = 3;\n  optional string soldToCustomer = 4;\n  optional string targetSystem = 5;\n}\n"
}

Put altogether we have this request to send a message:

curl --location --request POST 'http://{RestProxyHost}:{RestProxyPort}/topics/salesorder_topic' \
--header 'Content-Type: application/vnd.kafka.protobuf.v2+json' \
--data-raw '{
  "key_schema": null,
  "value_schema_id": 41,
  "records": [
    {
      "key": null,
      "value": {
      	"orderId": 1,
      	"orderType": "SERVICE",
      	"plant": 5000,
      	"soldToCustomer": "2000",
      	"targetSystem": "SYS1"
      }
    }
  ]
}'

Noticed the host and port to send requests are different for the REST Proxy and the Schema Registry.

Consume Message

It is not a single endpoint request as the producer. It has to follow steps in order to consume messages from the Kafka topic.

1.       Create a consumer instance

2.       Subscribe to a topic or list of topics.

3.       Consume messages

Create a consumer instance

This is a stateful request, the consumer instance is removed after 5 minutes idle, and the output of “base_uri” is used for the next steps.

The endpoint

POST / consumers /{group name}

Request

curl --location --request POST 'http://{RestProxyHost}:{RestProxyPort}/consumers/consumergroup' \
--header 'Content-Type: application/vnd.kafka.v2+json' \
--data-raw '{
  "name": "my_consumer_proto",
  "format": "protobuf"
}'

Notice the format value of “protobuf” is required as we are using Protobuf schema. The value is “avro” and “jsonschema” for Avro and JSON schema accordingly.

Example Response

{
    "instance_id": "my_consumer_proto",
    "base_uri": "http://kafkaproxy.sample/consumers/consumergroup/instances/my_consumer_proto"
}

Subscribe to a topic

The endpoint

{value fo base_uri from the previous step}/subscription

Request

curl --location --request POST 'http://kafkaproxy.sample/consumers/consumergroup/instances/my_consumer_proto/subscription' \
--header 'Content-Type: application/vnd.kafka.protobuf.v2+json' \
--data-raw '{
  "topics": [
    "salesorder_topic"
  ]
}'

Notice at Content-Type and topic name, this is topic with Protobuf schema that we want to consume messages.

Consume messages

The endpoint

{value fo base_uri from the previous step}/records

Accept header

Accept: application/vnd.kafka.protobuf.v2+json

Request

curl --location --request GET 'http://kafkaproxy.sample/consumers/consumergroup/instances/my_consumer_proto/records?timeout=3000&max_bytes=300000' \
--header 'Accept: application/vnd.kafka.protobuf.v2+json'

Notice at Accept header, when we created a group instance from the previous step with the format “protobuf” then we have to specify which content type we will accept for the response message.

Example Response

[
    {
        "topic": "salesorder_topic",
        "key": null,
        "value": {
            "orderId": 6,
            "orderType": "SERVICE",
            "plant": 5000,
            "soldToCustomer": "2000",
            "targetSystem": "SYS1"
        },
        "partition": 0,
        "offset": 1
    }
]

Conclusion

The REST Proxy API is a wrapper from the Java library and it is useful to test sending and receiving messages from Kafka. Still, it is not easy to use as we see it takes a couple of steps to receive messages and the group instance is expired every 5 minutes.

The same steps we can use for other content types such as Avro, JSON schema, all we need is to replace content type appropriately.

References

Confluent REST Proxy API reference
https://docs.confluent.io/current/kafka-rest/api.html

There are many databases on the market, but most of them are categorized into three data models relational, document, and graph. Data models and characteristics of the application to store and retrieve data are the factors to choose databases, and we do not compare which one is better. Very clear that the document model is the best for use cases where data is self-contained documents and no relationships of many-to-one or many-to-many between documents. The graph model is the opposite, and it targets to use cases where the data relationship is more complicated.

Neo4j is the most popular graph database now. In this blog, we use it to implement two functionalities as the diagram below.

-          Find related products: like most of the e-commerce websites, whenever a user views a product, he will see a list of related products that tempt him to buy more. The logic to find related products is different from each website, and here we are going to find products that were sold together with this product in orders that bought by any customers.

-          Create sales order: when a user makes an order, the e-commerce website sends it to the back office to process, then the sales order may be forwarded to another app to handle. In this case, we will send the sales order to the Neo4j database for storage and doing predictable tasks such as “find the related products.”

We use Advantco Neo4j Adapter to connect with Neo4j database. It supports native Cypher statements, which can store and retrieve data directly from the database, and it also uses their owner schema on top of Neo4j API to store and retrieve data. We will explore these two features throughout this implementation.

Find related products

We find products that were ordered together with this product in the past orders to recommend to customers.

The interface provides a REST API endpoint for the e-commerce website to look up the related products based on the viewing product. We won’t detail the REST channel configuration here but will express the request body later.

This interface we use the native Cypher statement feature of the Advantco Neo4j adapter, so no need for any message mapping. All we need is a Cypher query statement to get the data we want. The consumer of the REST API will send out the Cypher query as the request payload, and the adapter sends it to the Neo4j database and responds the result back to the consumer.

The simple channel configuration to work with Neoj4 database

There is no specific operation such as storage or retrieval in the channel configuration, only the connectivity information. Notice that we use the message protocol of “Native Neo4j Cypher Statement”, so the Cypher query that sent from the consumer will be wrapped to the appropriate Neo4j API, and the adapter takes care of this step.

There are other authentication methods from the basic one, such as Kerberos, LDAP, and Custom.

We can use any REST adapter or HTTP adapter to expose an endpoint API with the HTTP POST method as below.

https://server:port/AdvantcoRESTAdapter/RESTServlet/neo4j/api/v1/cypherquery

The Cypher query to get related products. We can use any as long as it returns the expected result based on our data model. In this case, we are viewing a product with ID 4 on the app, and the app looks for products that related to the product with ID 4.

match(p:Product {productID:"4"})<-[:orders]-(o:order), (o)-[:orders]->(rp:Product) where rp.categoryID = p.categoryID return rp

The sample of using the exposed API with CURL. It simulates the consumer which is the e-commer website. We just past the Cypher query as the request body.

curl --location --request POST 'server:port/AdvantcoRESTAdapter/RESTServlet/neo4j/api/v1/cypherquery' \
    --header 'Content-Type: application/json' \
    --header 'Content-Type: text/plain' \
    --data-raw 'match(p:Product {productID:"4"})<-[:orders]-(o:order), (o)-[:orders]->(pp:Product) where pp.categoryID = p.categoryID return pp'

Create Sales order

We do not focus on how the e-commerce website sends the order to the back-office, but we will go through the implementation that the back-office sends the sales order to the Neo4j database.

The previous interface of finding related products we don’t use message mapping by using the native Cypher query, we can see how fast to implement an interface with that way. But within this interface, we create a sales order from an IDOC, so we need to transform an IDOC structure to a structure that could accept by the Neo4j database, it is where we use the Advantco Schema on top of Neo4j API.

Let say we have a data model as below. We will use the Advantco Workbench to generate the schema needed for the mapping.

Base on the data model, we will create a new Order node and two edges (data relationships) of PURCHASED and ORDERS.

If you are from the data model perspective, you may ask why we need a schema for a schema-free database. It’s because we need the source and target data structure to do the data transformation in the middleware. In case we need more dynamic on data structure, we can refer to the way we implement “Find related products” to use native Cypher statements.

The sample mapping from IDOC to the schema

The payload looks like after the mapping

<?xml version="1.0" encoding="UTF-8"?>
<multiRequest>
   <request>
      <operation>MERGE</operation>
      <Order>
         <ID>orderID</ID>
         <LABEL>Order</LABEL>
         <propertiesToReturn>id</propertiesToReturn>
         <properties>
            <customerID type="string">match (c:Customer {customerID: &quot;LETSS&quot;}) return c.customerID</customerID>
            <orderID type="string">0003365889</orderID>
            <orderDate type="string">2020-03-24</orderDate>
         </properties>
      </Order>
      <PURCHASED>
         <START_ID>match (c:Customer {customerID: &quot;LETSS&quot;}) return c</START_ID>
         <END_ID>0003365889</END_ID>
         <TYPE>PURCHASED</TYPE>
      </PURCHASED>
      <ORDERS>
         <START_ID>0003365889</START_ID>
         <END_ID>match (p:Product) where p.productID in [&quot;1&quot;] return p</END_ID>
         <TYPE>ORDERS</TYPE>
         <properties>
            <unitPrice type="double">1719.0010</unitPrice>
            <productID type="string">1</productID>
            <quantity type="int">10</quantity>
         </properties>
      </ORDERS>
      <ORDERS>
         <START_ID>0003365889</START_ID>
         <END_ID>match (p:Product) where p.productID in [&quot;2&quot;] return p</END_ID>
         <TYPE>ORDERS</TYPE>
         <properties>
            <unitPrice type="double">020</unitPrice>
            <productID type="string">2</productID>
            <quantity type="int">20</quantity>
         </properties>
      </ORDERS>
      <ORDERS>
         <START_ID>0003365889</START_ID>
         <END_ID>match (p:Product) where p.productID in [&quot;3&quot;] return p</END_ID>
         <TYPE>ORDERS</TYPE>
         <properties>
            <unitPrice type="double">0.30</unitPrice>
            <productID type="string">3</productID>
            <quantity type="int">30</quantity>
         </properties>
      </ORDERS>
      <ORDERS>
         <START_ID>0003365889</START_ID>
         <END_ID>match (p:Product) where p.productID in [&quot;4&quot;] return p</END_ID>
         <TYPE>ORDERS</TYPE>
         <properties>
            <unitPrice type="double">0.40</unitPrice>
            <productID type="string">4</productID>
            <quantity type="int">40</quantity>
         </properties>
      </ORDERS>
   </request>
</multiRequest>

The channel configuration is not much different from the previous except the message protocols of Neo4j. The payload itself tells what to do (CREATE/UPDATE/DELETE/QUERY) on which nodes and edges, no need other specify on the channel configuration unless we want to customize different settings.

Note that we are using XML payload here, but the adapter also supports other payload types such as CSV and JSON.

Conclusion

It’s fast and easy to fulfill the requirement of integration with the Neo4j database on the SAP platform with the Advantco adapter.

From the integration perspective, it’s good to know the Cypher statement to either using native Cypher or using Advantco Schema to manipulate data through the adapter.

References

Cypher statement
https://neo4j.com/docs/cypher-manual/current/introduction/

Free Online Neo4j sandbox to play with graph database
https://neo4j.com/sandbox/

Advantco Neo4j adapter
https://advantco.com/product/adapter/neo4j

Doing data integration with Salesforce is not simple. We have to understand API objects which are analogous to database tables and KSQL to extract data from Salesforce, and different types of API set and when to use appropriately, and the concept of External Id… and limit of API calls.

With the Advantco Salesforce Adapter, we simplify the connectivity and transparently consume different API sets, so it makes the implementation more straightforward and faster.

The problem

Trying to make multiple API calls for a “simple” task and get exceeded API request limit?

Let examine an example of the sales order. SAP sends an sales order with a header and two order lines. It is not a simple example if we don’t want to say it’s a challenge in integration without adapter support.

We manually take care of the access token for every subsequence request calls to Salesforce by using Netweaver BPM or Java Mapping, and manually create mapping schema for each Salesforce object for REST/SOAP API calls…

To create an order header and several order lines, we need at least two requests to Salesforce. The first is to create the order header. In the second request, it looks up the created order header and then creates order lines. So we have to make sure these two requests execute in the right sequence, no other way.

The consequence of this restriction makes orders queue up when one of them failed in transmission to Salesforce.

In high throughput time, we may get the error of exceeding the API request limit if the number of requests is not managed well.

The solution

With the Advantco Salesforce adapter, the adapter supports session reuse to reduce the number of login call to Salesforce. So we maintain the connection and make a subsequent request in a single channel. It makes the integration flow clean, neither BPM needed nor an extra Java mapping to maintain the access token.

The choice of different types of API set is not a problem now. The adapter will do it. It’ll automatically select the appropriate API according to the feature we are using.

The schema is generated from the Advantco Salesforce Workbench, we don’t cover it in this topic, but it provides a ton of features such as Schema generation, SOQL editor, APEX client generator, Outbound Messaging (OBM)…

The implementation is faster and easier. But the queue up issue still.

You may know the Composite API. It helps to improve the performance by reducing round trip requests to Salesforce and minimize the number of API calls by batching up multiple calls to a single one.

With the Composite API support, we combine order header and order lines to a single request. It solves the problem. We don’t have to maintain the sequence of separate requests as earlier, so no more queue up and the transaction of each order is free from the other.

More important, it’s a lot easier to have a consistent transaction with the Composite request. Each batch of the Composite request can have up to 25 subrequests, and the batch is rolled back when an error occurs while processing a subrequest. It’s an optional configuration from the adapter.

The channel configuration using the Composite API.

The sample payload of the Composite request. The only difference from the payload of another API set is the field “referenceId”.

Look up the order header Id with the syntax “@{refQueryOrder.records[0].Id}”

The sample of message mapping.

Conclusion

So it’s straightforward to implement an integration scenario of sales order with the Composite API and the Advantco Salesforce Adapter.

In the real scenario, we may need to delete unused order lines along with adding new order lines and updating the existing lines. The sample payload we are using the operation UPSERT it means it already takes care of adding and updating existing, we need to add another subrequest of OrderDetail with an operation DELETE to delete unused order lines.

References

Advantco Salesforce Adapter
https://advantco.com/product/adapter/sfdc

Salesforce Composite Resource
https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_composite.htm

Confluent Schema Registry stores Avro schemas for Kafka producer and consumer so that producers write data with a schema that can be read by consumers even as producers and consumers evolve their using schema.

In this post, I am not going to explain what and how those tech terms work in theory. Instead, we will focus on the practical implementation in SAP PO. It’s almost identical to the cloud version of SAP CPI.

Our scenario is every time SAP sends a customer master (DEBMAS) to a third party it will send a copy to Kafka for real-time analytics or act as a middle stream for other apps. We use the Advantco Kafka adapter here.

Schema Evolution

To get up to speed in case you are not familiar with this subject, read the following paragraphs from the Confluent website to understand Avro schema and Confluent Schema Registry.

“An important aspect of data management is schema evolution. After the initial schema is defined, applications may need to evolve it over time. When this happens, it’s critical for the downstream consumers to be able to handle data encoded with both the old and the new schema seamlessly...

When using Avro, one of the most important things is to manage its schemas and consider how those schemas should evolve. Confluent Schema Registry is built for exactly that purpose.”

Prerequisites

Let start by creating Kafka stuff before jumping on the channel configuration.

Create a Kafka topic as the command below, or use any UI tool such as Confluent Control Center to create one.

./bin/kafka-topics --create --zookeeper localhost:2181   --replication-factor 1 --partitions 1 --topic debmas07_avro

Assume we have an Avro schema ready. If not we can use any online tools to generate one from an XML of the IDOC DEBMAS07 and make some changes if needed. To quickly have an Avro schema for this sample, I just simply use the Advantco Kafka Workbench to convert the XML payload to JSON and then use this online tool to generate an Arvo schema from the JSON.

Use Schema Registry API to upload the Avro schema to the Schema Registry, with a subject name debmas07_avro-value. Or any UI tool such as Advantco Kafka Workbench.

Create a stream from the created topic and schema. It’ll be used later for the real-time queries (KSQL). Run the command below after logging into the KSQL server.

CREATE STREAM sdebmas07_avro \
(KUNNR VARCHAR, \
NAME1 VARCHAR, \
STRAS VARCHAR, \
ORT01 VARCHAR, \
LAND1 VARCHAR) \
WITH (kafka_topic='debmas07_avro', value_format='AVRO', value_avro_schema_full_name='com.company.avro.Debmas07');

Producer channel

Assume we have an ICO that SAP sends an IDOC DEBMAS07 to a receiver channel of Kafka adapter.

The configuration of the receiver channel to produce messages to the Kafka topic. All messages will be converted to JSON and then serialize to Avro before sending it to Kafka broker. The Avro schema is stored on the Confluent Schema Registry and referencing to a schema by subject name and version.

Convert the XML payload to JSON format and store the only segment of E1KNA1M.

Consumer channel

Assume we have another ICO that consumes Kafka messages from the Kafka sender adapter and forward it to a receiver adapter, such as File.

Versions of Arvo schema can be the same or different on the sender and receiver channels. It depends on the Compatibility Type using in the Confluent Schema Registry. Here we use the BACKWARD compatibility type and the new schema version (version 2) to consume messages while the producer is still using the last version (version 1).

Let make a new version of the schema via the Advantco Kafka Workbench by edit the current version – delete field SEGMENT as the screenshot below and save it. So we have a new version 2 is different from the version 1 only field SEGMENT.

The Kafka sender channel consumes messages from the Kafka topic, it deserializes the message payload from the Avro schema which was used to serialize the message but in a new version.

So if we look at the output data of the interface we will not see field “SEGMENT” according to version 2 of the schema.

KSQL channel

KSQL is a stream engine that running SQL statements for Kafka. With powerful provided from KSQL, we easily query data in real-time.

In this example, we are querying customer who is from the US.

Assume we have another ICO that the Kafka sender adapter executes a KSQL query and a File adapter to receive the output.

We already have a stream data pipeline created above, and this is the channel configuration.

KSQL query.

SELECT KUNNR   as ID, NAME1 as Name, STRAS as BillingStreet, ORT01 as BillingCity, LAND1 as   BillingCountry FROM sdebmas07_avro WHERE LAND1 = 'US' EMIT CHANGES;

We don’t see any Arvo configuration on the channel because the stream sdebmas07_avro (from FROM clause) was created by reading Avro-formatted data.

Conclusion

Avro is a cross-languages serialization of data using a schema. It’s widely used in Kafka to serialize data between apps that developed in different platforms. And with the support from the adapter and the Confluent Schema Registry, we don’t have to write any single line of code to exchange data with other apps, as well as a central place to maintain the schema that developed by a team and reusing by other.

The Advantco Kafka adapter is supporting in two SAP platforms on-prems (PI/PO) and cloud (CPI). There is no difference in functionality.

References

Schema Evolution and Compatibility
https://docs.confluent.io/current/schema-registry/avro.html

Advantco Kafka Adapter
https://www.advantco.com/product/adapter/apache-kafka-adapter-for-sap-netweaver-pipo

Apache Avro
https://avro.apache.org/docs/current/

In this post, we are going to use 2 different clients to connect the Kafka broker with 2-way SSL. We will use Advantco Kafka adapter as well as Kafka console to produce and consume messages from the broker.

With 2-way SSL, here is how a client requests a resource in an encrypted channel:

1.       A client requests a protected topic from the broker
2.       The broker presents its identity by a certificate
3.       The client verifies the broker certificate
4.       In the other way, the client also sends its certificate to the broker for the verification
5.       The broker verifies the client certificate
6.       If success, the broker grants access to the topic to the client

In this post we just focus on how to setup on client side, and we assume the broker is set up with SSL/TLS and ready to use. We will need this information from the broker for the connectivity:

a.       The port that the broker listening for SSL (SSL://{port} or SASL_SSL://{port}).
b.       The certificate for the server authentication. It could be exported from the truststore (server.truststore.jks) or the keystore (server.keystore.jks). Let say a cluster with multiple brokers and we have a different keystore for each broker, but the CA Root certificate to sign the CSR for each keystore/broker is the same. Let name it as caroot_godaddy.crt for reference later.
c.       The client authentication flag by setting ssl.client.auth has to be “requested” or “required”.

We are using terms “keystore” and “truststore” frequently throughout the article. Let explain it first, “keystore” is a repository/file that stores private-public keys and certificates, and “truststore” is a repository/file that stores only certificates.

Now we have the certificate from the broker (caroot_godaddy.crt) for the server authentication (aka 1-way SSL). For the client authentication (second way SSL), we need a keystore which includes public-private keys and a signed certificate (self-signed or from a trusted CA) and a CA Root certificate that signed the certificate.

Create a keytstore and a CA Root certificate

The CA Root certificate is to sign the certificate signing request (CSR), and we can generate it to sign the certificate (self-signed) or we can export it from a trusted CA.

Let create a keystore for the client.

Create a client keystore

keytool -genkey -keystore client.keystore.jks -validity 3650 -storepass "password1" -keypass "password1" -dname "CN=client1" -alias client1 -storetype pkcs12

Create a client cert sign request (CSR)

keytool -keystore client.keystore.jks -certreq -file client-cert-sign-request -alias client1 -storepass "password1" -keypass "password1"

The CSR file could be signed by a trusted CA or by self-signed.

Here are steps if we want to sign the certificate by ourselves

openssl req -new -x509 -keyout ca-key -out ca-cert -days 3650

openssl x509 -req -CA ca-cert -CAkey ca-key -in client-cert-sign-request -out client-cert-signed -days 3650 -CAcreateserial -passin pass:password1

cat ca-cert client-cert-signed > completeChain.crt

Now we will import the certificate chain to the keystore.

keytool -keystore client.keystore.jks -import -file completeChain.crt -alias client1 -storepass "password1" -keypass "password1" -noprompt

The keystore is ready to use. The last step is to export the CA Root certificate.

If we signed the certificate by ourselves then the ca-cert is the CA Root certificate.

We are using KeyStore Explorer to export the CA Root certificate.

Import the CA Root certificate to Kafka broker

This step should be performed by Kafka team. We send the CA Root certificate file from the previous step and ask them to import it to the broker truststore.

Assume the broker truststore file name is server.truststore.jks

keytool -keystore server.truststore.jks -alias carootsapcpip0520 -import -file carootsapcpip0520.crt -storepass "{password of truststore}" -noprompt

This step is not always required. Like in a corporation we are using the same trusted CA to sign certificates for multiple system including Kafka brokers, then the CA Root certificate is already in the truststore.

Make a connection with Advantco Kafka Adapter

There are different versions for 2 platforms SAP CPI/HCI and PI/PO, but the adapter configuration is identical.

The 2-way SSL can be used for none authentication mode as long as other authentication modes such as Kerberos, Plain, SCRAM..

We have a keystore (client.keystore.jks)and a certificate (caroot_godaddy.crt)from the previous steps, let import it to the Key Storage for referencing in the channel configuration later.

Java KeyStore (JKS) format of the keystore is not supported by PI/PO Key Storage, we have to convert it to P12 format.

Convert JKS to P12

keytool -importkeystore -srckeystore client.keystore.jks -destkeystore client.keystore.p12 -deststoretype pkcs12

Key Storage View SAP_KAFKA_CLIENT

Similar with SAP CPI/HCI, create 2 entries one for Keystore and one for Certificate. The cloud platform supports JKS, so we don’t have to convert the keystore to P12 format.

Here is an example of 2-way SSL with Kerberos

Test the connectivity with Kafka console

The best way to test 2-way SSL is using Kafka console, we don’t have to write any line of code to test it.

Simply download Kafka from Apache Kafka website to the client, it includes kafka-console-producer and kafka-console-consumer in bin directory. We will use one of it to test the connectivity.

To use the console we have to create 2 things:

Wrapper the server certificate (caroot_godaddy.crt) to a client truststore

keytool -keystore client.truststore.jks -alias CARoot -import -file caroot_godaddy.crt -storepass "password1" -noprompt

Create a client property file client-ssl.properties and copy it to kafka bin directory, should have following lines:

security.protocol=SSL
ssl.truststore.location=client.truststore.jks
ssl.truststore.password=password1
ssl.endpoint.identification.algorithm=
ssl.keystore.location=client.keystore.jks
ssl.keystore.password=password1
ssl.key.password=password1

It’s ready to use the kafka console producer to produce a test message to a topic

./bin/kafka-console-producer --broker-list {broker name}:{port} --producer.config client-ssl.properties –topic {topic name}

We are done. Hope you find it useful.

In this post, we will go through a scenario where we use Google Cloud Platform (GCP) serverless services to archive Event-Driven model.

The scenario is the Ink Replenishment program, whenever a user buys a printer at a store or online the system will send the user an email to register to the program. If the user decided to opt-in to the program and then the user will automatically receive a suggestion to buy new Ink whenever the ink low. So the user doesn’t have to look up for the Ink code of the printer in order to buy a new one, she just responses back Yes or No to the system to buy a new Ink.

Once the Yes response sent, the system finds the store who sold the printer based on the serial number and forward the order to the store to fulfill it. We will address this order scenario in the event-driven model in this blog.

Let says we have a chain of Command and Event as the diagram below.

The very first command is from the customer, she responses YES to trigger the order workflow from her computer/Phone, and the last command is to fulfill the order and it will be taken care from the external stores. There are 2 other commands need to implement in the internal system are Package Order and Send Order.

Before we go to the detail of the implementation, let have a quick explanation about Event the diagram above. We are using Google Cloud Pub/Sub for messaging events, and each event is generated with a specific data object after the command executed. (That is the reason why we don’t see any data object in the diagram except a serial of commands and events)

An example for the event Order Submitted: once the customer confirmed to order the new Inks, the application on the customer laptop/phone will trigger an event and push it to a topic in Cloud Pub/Sub, and then the event will trigger another command which is subscribing to the topic.

{
	"customerId": "0000001234",
	"serial": "serial number of the printer",
	"date": "2019-09-06T14:38:00.000",
	"items": [
		{
			"code": "64",
			"color": "Y",
			"inklevel": 24
		},
		{
			"code": "64",
			"color": "B",
			"inklevel": 10
		}
	]
}

SAP CPI and Advantco GCP Adapter

Now it’s time to think about the implementation commands of Package Order and Send Order. We can use Google Cloud Functions to respond to the events from Cloud Pub/Sub to process, transform and enrich data. So, Cloud Functions is a good glue to connect point to point, specifically as a command in our model to consume data from an event and then generate a new event to trigger another command.

But besides the effort of coding to connect to the back-office such as SAP, Salesforce… which are not in GCP stack. We do have a better and faster option which is SAP CPI, it already has adapters (IDOC, ODATA, SFO…) to connect to those systems.

Look at the diagram below where Salesforce is the back office and SAP CPI connecting events with help from Advantco GCP adapter to consume and produce event messages from Google Cloud Pub/Sub.

Package Order

The flow of the command Package Order. It makes the flow so much simple with Advantco GCP adapter and SFO (Salesforce) adapter.

The GCP sender channel, it does some other stuff beside pulling event messages from Cloud Pub/Sub such as uncompressing GZIP message, converting JSON message to XML. It makes the flow clean and simple.

The connection configuration within the adapter supports Service Account and OAuth2, it can open multiple concurrent connections for a high volume of flowing messages.

Content Enricheris a very cool step from SAP CPI, it helps to combine/enrich two data sources to one. Yes, in a single step without Message Mapping. It’s a reason why we converted the SubmittedOrder from JSON to XML in the GCP sender channel above.

Source 1 - a submitted order

<?xml version="1.0" encoding="UTF-8"?>
<SubmitedOrder>
	<customerId>0000001238</customerId>
	<serial>1234554321</serial>
	<date>2019-09-06T14:38:00.000</date>
	<items>
		<item>
			<code>64</code>
			<color>Y</color>
			<inklevel>24</inklevel>
		</item>
		<item>
			<code>64</code>
			<color>B</color>
			<inklevel>10</inklevel>
		</item>
	</items>
</SubmitedOrder>
        	

Source 2 - customer and store info

<?xml version="1.0" encoding="UTF-8"?>
<queryResponse>
	<Customer__c>
		<id>...</id>
		..		
		<External_CustID__C>0000001238</External_CustID__C>
		<Store>
			<id>...</id>
			..
		</Store>
	</Customer__c>
</queryResponse>
            

Target - the combine of Source 1 and Source 2

<?xml version="1.0" encoding="UTF-8"?>
<SubmitedOrder>
	<customerId>0000001238</customerId>
	<Customer__c>
		<id>...</id>
		..		
		<External_CustID__C>0000001238</External_CustID__C>
		<Store>
			<id>...</id>
			..
		</Store>
	</Customer__c>
                    <serial>1234554321</serial>
	<date>2019-09-06T14:38:00.000</date>
	<items>
		<item>
			<code>64</code>
			<color>Y</color>
			<inklevel>24</inklevel>
		</item>
		<item>
			<code>64</code>
			<color>B</color>
			<inklevel>10</inklevel>
		</item>
	</items>
</SubmitedOrder>

The GCP Receiver channel, it creates a new event after the order packaged by producing an event message to Cloud Pub/Sub.

As an agreement in the order workflow, all event messages are stored in JSON format and compressed with GZIP. So, the GCP receiver channel will do these things to make it uniform.

Send Order

The flow of the command Send Order. Before sending out the order we store it in BigQuery - another serverless service from Google, then we can keep track of the order status as well as analyze and have a statistic in a later time.

Consuming and producing event messages from Cloud Pub/Sub using Advantco GCP adapter is similar steps above, with compress/uncompress GZIP and conversion between XML and JSON.

Beside the Cloud Pub/Sub the adapter also supports many other services such as BigQuery, Datastore, Spanner, Dataflow, Storage and Google Drive. Here is the list of services with authentication methods supported by the adapter.

We use BigQuery to store the order in the flow.

Summary

We did not go much detail on the implementation but we see many benefits from the Google serverless services in the project implementation, no worrying of server management, stability, scalability… we can start the project from day one.

And the most important is the Advantco GCP adapter played a key role in this project, it speeded up the implementation with the functions it offers.

References

Advantco GCP Adapter
https://advantco.com/product/adapter/google-cloud-platform-adapter-for-sap-pi-po

Cloud Pub/Sub
https://cloud.google.com/pubsub/

BigQuery
https://cloud.google.com/bigquery/