Delay in batch load in Sink:JDBC Application

[bhattpruthvi11]

Hi Team,

I have deployed two streams, definition "kafka | rtitconverter | jdbc" for each kafka topic, in Spring Cloud dataflow server where kafka and jdbc are inbuilt applications and rtitconverter is custom built processor.

From kafka broker:
Topic: A, we are processing 4 Billion records per day and there is no lag in data load from JDBC. (stream 1)
Topic: B, we are processing 141 Billion data per day. But there is delay of 6 hrs in data load from JDBC. (stream 2)

Below are the jdbc properties for Topic: B input used in jdbc application:

app.jdbc.batch-size=2000
app.jdbc.idle-timeout=60000
app.jdbc.columns=subscriber_id:payload.subscriber_id,start_time:T(java.time.LocalDateTime).parse(payload.R_start_time),end_time:T(java.time.LocalDateTime).parse(payload.R_end_time),subscriber_msisdn:payload.subscriber_msisdn,imei:payload.IMEI,sv:payload.R_SV,tac:payload.TAC,cell_mcc:payload.cell_mcc,cell_mnc:payload.cell_mnc,cell_id:payload.R_CellId,celltype_name:payload.celltype_name,source_host_ip_address:payload.source_host_ip_address,destination_host_ip_address:payload.destination_host_ip_address,ctrlplane_trans_status_name:payload.controlplane_transaction_status_name,message_protocol_type_code:payload.message_protocol_type_code,response_code:payload.response_code,response_description:payload.response_description,MOVOICEATTEMPT:payload.Movoiceattempt,MOVOICECONNECT:payload.Movoiceconnect,MOVOICEDROPPED:payload.Movoicedropped,MOVOICEDROPPEDNETWORK:payload.Movoicedroppednetwork,MOVOICEDROPPEDUSER:payload.Movoicedroppeduser,MOVOICEDURATION:payload.Movoiceduration,MOVOICEDURATIONCOUNT:payload.Movoicedurationcount,MOVOICESETUPFAIL:payload.Movoicesetupfail,MOVOICESETUPFAILNETWORK:payload.Movoicesetupfailnetwork,MOVOICESETUPFAILUSER:payload.Movoicesetupfailuser,MOVOICESETUPTIME:payload.Movoicesetuptime,MOVOICESETUPTIMECOUNT:payload.Movoicesetuptimecount,MOVOICESUCCESS:payload.Movoicesuccess,MTVOICEATTEMPT:payload.Mtvoiceattempt,MTVOICECONNECT:payload.Mtvoiceconnect,MTVOICEDROPPED:payload.Mtvoicedropped,MTVOICEDROPPEDNETWORK:payload.Mtvoicedroppednetwork,MTVOICEDROPPEDUSER:payload.Mtvoicedroppeduser,MTVOICEDURATION:payload.Mtvoiceduration,MTVOICEDURATIONCOUNT:payload.Mtvoicedurationcount,MTVOICESETUPFAIL:payload.Mtvoicesetupfail,MTVOICESETUPFAILNETWORK:payload.Mtvoicesetupfailnetwork,MTVOICESETUPFAILUSER:payload.Mtvoicesetupfailuser,MTVOICESETUPTIME:payload.Mtvoicesetuptime,MTVOICESETUPTIMECOUNT:payload.Mtvoicesetuptimecount,MTVOICESUCCESS:payload.Mtvoicesuccess,VOICEATTEMPT:payload.Voiceattempt,VOICECONNECT:payload.Voiceconnect,VOICEDROPPED:payload.Voicedropped,VOICEDROPPEDNETWORK:payload.Voicedroppednetwork,VOICEDROPPEDUSER:payload.Voicedroppeduser,VOICEDURATION:payload.Voiceduration,VOICESETUPFAIL:payload.Voicesetupfail,VOICESETUPFAILNETWORK:payload.Voicesetupfailnetwork,VOICESETUPFAILUSER:payload.Voicesetupfailuser,VOICESETUPTIME:payload.Voicesetuptime,VOICESETUPTIMECOUNT:payload.Voicesetuptimecount,VOICESUCCESS:payload.Voicesuccess,MOVOICESETUPSUCCESS:payload.Movoicesetupsuccess,MTVOICESETUPSUCCESS:payload.Mtvoicesetupsuccess,VOICESETUPSUCCESS:payload.Voicesetupsuccess,LOAD_SERVER:'S09',DEVICE_NAME:payload.device_name
app.jdbc.spring.cloud.stream.bindings.input.content-type=application/json
app.jdbc.spring.datasource.driver-class-name=oracle.jdbc.OracleDriver
app.jdbc.spring.datasource.password=test
app.jdbc.spring.datasource.url=jdbc:oracle:thin:@//dev_env:33001/dev.test
app.jdbc.spring.datasource.username=test
app.jdbc.table-name=topic_b

Can you please help to identify and solve the delay in process load? Let me know if you require more details.

Thank you in advance