Unable to receive messages from a queue in Enterprise Messaging and consume in Data Intelligence

[former_member513362]

Hello colleagues,

I have created a CPEM (Cloud Platform Enterprise Messaging) and DWC (Data Warehouse Cloud) Connection in SAP Data Intelligence. I was able to test DWC connection and it works fine but CPEM connection test is not supported.

I created a pipeline in ML scenario manager using SAP CP EM Consumer operator and trying to insert data into one of the database tables in DWC.

But, this scenario is not working and when I try to run the pipeline, it gives a unreadable lengthy error.

I have few doubts here :
1) What should be provided in the host details while creating the CPEM connection? For me, the host looks something like below, is this ok?

enterprise-messaging-messaging-gateway.cfapps.sap.hana.ondemand.com

2) In client id and secret should be of Enterprise Messaging service instance or service key?

3) Only if the pipeline runs green, does it mean that it is deployed and ready to work? How I can troubleshoot the problem?

Could someone provide me any hints/suggestions here ?

Thanks & Regards,

Jhansi

[kai-michael_roesner · Accepted solution]

Hi Jhansi,

the host you provided in your CPEM connection looks okay. You should take the client ID and secret from the service key you created for your Enterprise Messaging instance as described here.

While your pipeline is running it is shown with a blue "running" status in Pipeline Modeller. The green "completed" status indicates that the pipeline has successfully finished:

You can download a diagnostics zip file by clicking the "Download Diagnostic Information" button as indicated in the above screenshot of the "SAP CP EM Production Example" pipeline which you can find in the "SAP Integration" category of graphs.

You can find information about the structure and content of this archive here. You want to look at the "logs-<podname>.txt" file in the "<graph-source>-<handle>" folder.

[former_member513362 · Accepted solution]

Hello Roesner,

Thank you very much for providing the useful information. I am able to receive EM messages in DI now and monitor in the payload monitor.

I am trying with the next part now, i.e. to insert/upsert the received payload data into DWC DB tables.

Regards,

Jhansi

[former_member513362 · Accepted solution]

Hello Roesner,

I am looking for some hints/suggestion regarding SAP HANA Client Operator - JSON input format.

I am trying to insert data into a predefined table using SAP HANA Client Operator. The message arrives at port data without the hana.preparedStatement attribute. The message body is in JSON input format. But, the pipeline always fails with the below error message:
{"message.error":true,"message.response.error":"failed to parse JSON array from input: json: cannot unmarshal object into Go value of type []interface {}"}failed to parse JSON array from input: json: cannot unmarshal object into Go value of type []interface {}.

I tried with CSV input format and the same works fine i.e. data is getting inserted to HANA DB tables. But, I want to achieve this with JSON input instead of CSV. Would you be able to provide an example of the expected JSON ?

Thanks & Regards,

Jhansi

[kai-michael_roesner · Accepted solution]

Hi Jhansi,

when using JSON format for the data input port the data must be a JSON array. Each of its elements can be either:

• a JSON array representing a table row (with elements in the same order as specified in the "Table columns" config) e.g.
  

  [
      ["1", 1, 2, 3 ],
      ["2", 2, 4, 6 ],
      ["3", 1, 4, 8 ]
  ]
  

  or
• a JSON object representing a table row (with keys equal to the column names specified in the "Table columns" config) e.g.
  

  [
      {"id": "1", "col1": 1, "col2": 2, "col3": 3 },
      {"id": "2", "col1": 2, "col2": 4, "col3": 6 },
      {"id": "3", "col1": 1, "col2": 4, "col3": 8 }
  ]

You can see an example of how that works in the "File-to-DB Javascript" graph in the "Scenario Templates" category.

[former_member513362 · Accepted solution]

Hi Roesner,

Thanks for your inputs. I tried with the JSON format suggested by you and it worked 🙂

You referred to an example in the "File-to-DB Javascript" graph in your answer and I had checked this already but I was unable to find the input information from the pipeline or even run it. As when I run the pipeline it fails with the error message "Could not find connection with id CUSTDATAPROC_HANA."

Also, looks like most the configuration of example pipelines have global variables which needs to be populated at runtime. Due to which, I can check the operators used and all the static information but not really the input and output of the operators.

Do you have some suggestions or tips to use these example pipelines and learn from it better?

Thanks & Regards,

Jhansi

[kai-michael_roesner · Accepted solution]

Hi Jhansi,

please refer to the graph documentation. Open the graph in the Modeler (don't select any of the operators) and click the "Show Documentation" button at the top right corner of the Modeler screen. If you just want to see the data that is fed into the "SAP HANA Client" operator you can replace that operator by a "Terminal" or "Wiretap" operator. In this case you won't need to configure a HANA connection.

Best regards
Kai.

[former_member513362 · Accepted solution]

Thank you so much for your inputs, Kai. It really helped 🙂

I have one more question w.r.t Write HANA Table (com.sap.hana.writeTable) operator. I am using Dynamic ( from input) configuration mode where I want to insert/upsert data to different database tables in DWC based on the payload received .

Here, the problem is that when all the columns of the DB table are of type string then the insertion doesn't work but whenI changed one of the column from string to integer the pipeline starts to work. Could you help me understand if this is expected ? If not, how could I make this work?

Below are some details about my pipeline:

Error message while inserting ( if all columns are of type string)

[2021-02-24 13:01:51,000] {"error":{"input":{"Attributes":{"table":{"columns":[{"name":"WORK_PACKAGE_RUN","size":5000,"type":"NVARCHAR"},{"name":"PROCEDURE_RUN_EXTERNAL_ID","size":100,"type":"NVARCHAR"},{"name":"PROCEDURE_TYPE","size":100,"type":"NVARCHAR"},{"name":"PROCEDURE_RUN_NAME","size":100,"type":"NVARCHAR"},{"name":"PROCEDURE_EXTERNAL_ID","size":100,"type":"NVARCHAR"},{"name":"PROCEDURE_NAMESPACE","size":100,"type":"NVARCHAR"},{"name":"DESTINATION","size":100,"type":"NVARCHAR"},{"name":"PROCEDURE_INPUT_ID","size":100,"type":"NVARCHAR"}],"name":"WORKPACKAGE_PROCEDURE_RUN","version":1}},"Body":[["86eec615-85cd","285ec703-12ca","JR Test DI 132","ME-09-23","111","86eec615-85cd","CXT","982f4321-b182"]],"Encoding":"json"},"message":"expected message body to be an array, but found [][]string instead","source":{"operation":"writeTable","operator":"com.sap.hana.writeTable","port":"","process":"writehanatable111"},"version":1}}expected message body to be an array, but found [][]string instead

Below Payload doesn't work :

{

"tableData": {

"table": {

"version": 1,

"name": "WORKPACKAGE_PROCEDURE_RUN",

"columns": [{"name":"WORK_PACKAGE_RUN","type":"NVARCHAR","size":5000},{"name":"PROCEDURE_RUN_EXTERNAL_ID","type":"NVARCHAR","size":100},{"name":"PROCEDURE_TYPE","type":"NVARCHAR","size":100},{"name":"PROCEDURE_RUN_NAME","type":"NVARCHAR","size":100},{"name":"PROCEDURE_EXTERNAL_ID","type":"NVARCHAR","size":100},{"name":"PROCEDURE_NAMESPACE","type":"NVARCHAR","size":100},{"name":"DESTINATION","type":"NVARCHAR","size":100},{"name":"PROCEDURE_INPUT_ID","type":"NVARCHAR","size":100}]

}

},

"Encoding": "table",

"Body": [

[ "86eec615-85cd", "285ec703-12ca", "JR Test DI 132", "ME-09-23", "111", "86eec615-85cd", "CXT", "982f4321-b182" ]

]

}

Below Payload works :

{

"tableData": {

"table": {

"version": 1,

"name": "DI_WP_PROCEDURE_RUN",

"columns": [{"name":"WORK_PACKAGE_RUN","type":"NVARCHAR","size":5000},{"name":"PROCEDURE_RUN_EXTERNAL_ID","type":"NVARCHAR","size":100},{"name":"PROCEDURE_TYPE","type":"NVARCHAR","size":100},{"name":"PROCEDURE_RUN_NAME","type":"NVARCHAR","size":100},{"name":"PROCEDURE_EXTERNAL_ID","type":"INTEGER"},{"name":"PROCEDURE_NAMESPACE","type":"NVARCHAR","size":100},{"name":"DESTINATION","type":"NVARCHAR","size":100},{"name":"PROCEDURE_INPUT_ID","type":"NVARCHAR","size":100}]

}

},

"Encoding": "table",

"Body": [

[ "86eec615-85cd", 121, "JR Test DI 121", "ME-09-23", "121", "86eec615-85cd", "CXT", "982f4321-b182" ]

]

}

Any inputs here would be of great help! Thanks!

Regards,

Jhansi

[kai-michael_roesner · Accepted solution]

Jhansi, please don't use comments to ask follow-up questions that are unrelated to the original post. That way the community will not be able to benefit from the conversation. You should either post a new question, or else, if it's very specific you can also reach out to me via Slack or e-mail.

[former_member513362 · Accepted solution]

Hello Kai,

Sorry for this, I do understand. Will reach you directly over email.

Regards,

Jhansi