Jolt specification working for Json input but result showing null in Nifi putdatabaserecord processor

[#1]

Json input:

{
"schema": {
"type": "struct",
"fields": [
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "COUNTRY_ID"
},
{
"type": "string",
"optional": true,
"field": "COUNTRY_NAME"
},
{
"type": "string",
"optional": true,
"field": "REGION_ID"
}
],
"optional": true,
"name": "poc2.NIFITEST.COUNTRIES.Value",
"field": "before",
"version": null
},
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "COUNTRY_ID"
},
{
"type": "string",
"optional": true,
"field": "COUNTRY_NAME"
},
{
"type": "string",
"optional": true,
"field": "REGION_ID"
}
],
"optional": true,
"name": "poc2.NIFITEST.COUNTRIES.Value",
"field": "after",
"version": null
},
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "version",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": false,
"field": "connector",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": false,
"field": "name",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "int64",
"optional": false,
"field": "ts_ms",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": true,
"field": "snapshot",
"name": "io.debezium.data.Enum",
"version": 1,
"parameters": {
"allowed": "true,last,false,incremental"
},
"default": "false"
},
{
"type": "string",
"optional": false,
"field": "db",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": true,
"field": "sequence",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": false,
"field": "schema",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": false,
"field": "table",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": true,
"field": "txId",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": true,
"field": "scn",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": true,
"field": "commit_scn",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": true,
"field": "lcr_position",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": true,
"field": "rs_id",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "int64",
"optional": true,
"field": "ssn",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "int32",
"optional": true,
"field": "redo_thread",
"name": null,
"version": null,
"parameters": null,
"default": null
},
{
"type": "string",
"optional": true,
"field": "user_name",
"name": null,
"version": null,
"parameters": null,
"default": null
}
],
"optional": false,
"name": "io.debezium.connector.oracle.Source",
"field": "source",
"version": null
},
{
"type": "string",
"fields": null,
"optional": false,
"name": null,
"field": "op",
"version": null
},
{
"type": "int64",
"fields": null,
"optional": true,
"name": null,
"field": "ts_ms",
"version": null
},
{
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "id"
},
{
"type": "int64",
"optional": false,
"field": "total_order"
},
{
"type": "int64",
"optional": false,
"field": "data_collection_order"
}
],
"optional": true,
"name": "event.block",
"field": "transaction",
"version": 1
}
],
"optional": false,
"name": "poc2.NIFITEST.COUNTRIES.Envelope",
"version": 1
},
"payload": {
"before": null,
"after": {
"COUNTRY_ID": "IT",
"COUNTRY_NAME": "Italy",
"REGION_ID": "1"
},
"source": {
"version": "2.4.0.Final",
"connector": "oracle",
"name": "poc2",
"ts_ms": 1699378780000,
"snapshot": "false",
"db": "TESTDB",
"sequence": null,
"schema": "NIFITEST",
"table": "COUNTRIES",
"txId": "030005002e290000",
"scn": "38041443",
"commit_scn": "38041466",
"lcr_position": null,
"rs_id": "0x000476.0002ce07.0010",
"ssn": 0,
"redo_thread": 1,
"user_name": "NIFITEST"
},
"op": "c",
"ts_ms": 1699378784139,
"transaction": null
}
}

Jolt Specification:

[
{
"operation": "shift",
"spec": {
"payload": {
"after": " "
}
}
}
]

PutDatabaseRecord showing null:

Source : Kafka Topics thru Debezium

Nifi Version 1.17

My Flow is: ConsumeKafkarecord_2.0--> Splitrecord-->EvaluateJsonPath(Gettable fromflowfile)-->RouteonAttribute-->Queryrecord-->JoltTransformJSON-->PutDatabaseRecord

Any solution please.

Appreciate your help.

Thanks,

Sree

[#2]

@CE Welcome to the Cloudera Community!

To help you get the best possible solution, I have tagged our NiFi experts @MattWho @SAMSAL  who may be able to assist you further.

Please keep us updated on your post, and we hope you find a satisfactory solution to your query.

[#3]

Hi @CE ,

Where exactly are you getting the null? are you getting it after the Jolt processor in the success relationship queue? I ran the same spec against simple GenerateFlowFile with the input you provided set in the CustomText property then use the Jolt processor with the provided spec and I got the expected output. Can you try that and see if it works? Can you also provide processor configuration screenshot ?

Thanks

[#4]

Hi @SAMSAL,

JoltTransformJSON-->PutDatabaseRecord in this two processor's success relationships showing null.

Thank you

[#5]

Also i have tested GenerateFlowfile with custom text its working. Problem is JoltTransformJSON-->PutDatabaseRecord in this two processor's success relationships showing null.

[#6]

Are you sure you are getting the correct input to the JoltTransformJSON processor? If you are able to test the spec using GenerateFlowFile and its working then something might be happening upstream that might be causing the input to change or dropped. I would check the EvaluateJsonPath  Destination property and make sure its not set flowfile-content, rather it should be set to flowfile-attribute. If you think everything is set correctly, can you take screenshot of all processors configurations in your reply? Thanks

[#7]

@CE Has the reply helped resolve your issue? If so, please mark the appropriate reply as the solution, as it will make it easier for others to find the answer in the future. If you are still experiencing the issue, can you provide the information @SAMSAL has requested? Thanks.