Kafka lag issue log line

municipal-services/ws-calculator/src/main/java/org/egov/wscalculation/consumer/DemandGenerationConsumer.java

@@ -289,8 +289,6 @@ private void generateDemandAndSendnotification(RequestInfo requestInfo, String t
 		Map<String, Object> masterMap = mDataService.loadMasterData(requestInfo,
 				tenantId);
 
-		log.info("connectionNos" + connectionNos.size());
-		log.info("connectionNos" + connectionNos);
 		log.info("dayStartTime:"+dayStartTime);
 		log.info("dayEndTime"+dayEndTime);
 
@@ -577,9 +575,12 @@ public void generateBulkDemandForULB(HashMap<Object, Object> messageData) {
 		String billingPeriod = bulkDemand.getBillingPeriod();
 		if (StringUtils.isEmpty(billingPeriod))
 			throw new CustomException("BILLING_PERIOD_PARSING_ISSUE", "Billing Period can not be empty!!");
-		log.info("CALL FROM TOPIC egov.generate.bulk.demand.manually.topic" );
+		log.info("CALL FROM TOPIC egov.generate.bulk.demand.manually.topic for tenantid:"
+				+bulkDemand.getTenantId()+" BillPeriod:"+billingPeriod+" Start Time:"+System.currentTimeMillis() );
 		generateDemandAndSendnotification(bulkDemand.getRequestInfo(), bulkDemand.getTenantId(), billingPeriod, billingMasterData,
 				isSendMessage, isManual);
+		log.info("System end time for Tenantid:"+bulkDemand.getTenantId()+" BillPeriod:"+billingPeriod+" end Time:"+System.currentTimeMillis());
+
 
 	}
 	@KafkaListener(topics = {
@@ -592,6 +593,8 @@ public void updateAddPenalty(HashMap<Object, Object> messageData) {
 	@KafkaListener(topics = {
 			"${ws.generate.demand.bulk}" })
 	public void generateDemandInBulkListner(HashMap<Object, Object> messageData) {
+		log.info("GOt call inn ws-gennerate-demand-bulk topic Started time:" +System.currentTimeMillis());
+		Long starttime= System.currentTimeMillis();
 		CalculationReq calculationReq= new CalculationReq();
 		Map<String, Object> masterMap = new HashMap<>();
 		String billingCycle ;
@@ -606,6 +609,10 @@ public void generateDemandInBulkListner(HashMap<Object, Object> messageData) {
 
 		log.info("got generate demand call for :"+calculationReq.getCalculationCriteria().get(0).getConnectionNo());
 		generateDemandInBulk(calculationReq,billingCycle,masterMap,isSendMessage,tenantId);
+		log.info("GOt call inn ws-gennerate-demand-bulk topic end time:" +System.currentTimeMillis());
+		Long enndtime=System.currentTimeMillis();
+		long diff= enndtime-starttime;
+		log.info("Time taken to process request for :"+calculationReq.getCalculationCriteria().get(0).getConnectionNo()+" is :"+diff/1000 +" secs");
 	}
 
 }

municipal-services/ws-calculator/src/main/java/org/egov/wscalculation/service/DemandService.java

@@ -1231,6 +1231,7 @@ public void generateBulkDemandForTenantId(BulkDemand bulkDemand) {
 		demandData.put("billingMasterData", billingMasterData);
 		demandData.put("bulkDemand", bulkDemand);
 		producer.push(config.getGenerateBulkDemandTopic(), demandData);
+		log.info("Pushed TO Genarate demand Bulk manual topic" +bulkDemand.getTenantId()+" Billing period "+bulkDemand.getBillingPeriod());
 //		generateBulkDemandForULB(billingMasterData, bulkDemand);
 	}
 

municipal-services/ws-calculator/src/main/java/org/egov/wscalculation/web/controller/CalculatorController.java

@@ -79,6 +79,7 @@ public ResponseEntity<DemandResponse> updateDemands(@RequestBody @Valid RequestI
 
 	@PostMapping("/_bulkDemand")
 	public ResponseEntity<BulkDemandResponse> bulkDemand(@Valid @RequestBody BulkDemand bulkDemand) {
+		log.info("got call to generatedemand for tenantid:"+bulkDemand.getTenantId()+" period:"+bulkDemand.getBillingPeriod());
 		wSCalculationService.generateBulkDemandForTenant(bulkDemand);
 		BulkDemandResponse response = BulkDemandResponse.builder().message("Bulk demand generation process stated, you will be notified shortly!")
 				.responseInfo(responseInfoFactory.createResponseInfoFromRequestInfo(bulkDemand.getRequestInfo(), true)).build();

municipal-services/ws-calculator/src/main/resources/application.properties

@@ -158,4 +158,8 @@ is.penalty.feature.enable= true
 egov.update.demand.add.penalty= egov-update-demand-add-penalty
 
 #GenareteDemndINBulk
-ws.generate.demand.bulk=ws-generate-demand-bulk
+ws.generate.demand.bulk=ws-generate-demand-bulk
+
+spring.kafka.consumer.session.timeout.ms=30000
+spring.kafka.consumer.heartbeat.interval.ms=10000
+spring.kafka.consumer.max.poll.interval.ms=600000