Kafka lag issue log line

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

@@ -274,23 +274,12 @@ private void generateDemandAndSendnotification(RequestInfo requestInfo, String t
 
 
 
-		Calendar previousFromDate = Calendar.getInstance();
-		Calendar previousToDate = Calendar.getInstance();
-
-		previousFromDate.setTimeInMillis(dayStartTime);
-		previousToDate.setTimeInMillis(dayEndTime);
 
-		previousFromDate.add(Calendar.MONTH, -1); //assuming billing cycle will be first day of month
-		previousToDate.add(Calendar.MONTH, -1); 
-		int max = previousToDate.getActualMaximum(Calendar.DAY_OF_MONTH);
-		previousToDate.set(Calendar.DAY_OF_MONTH, max);
 		String assessmentYear = estimationService.getAssessmentYear();
 		ArrayList<String> failedConnectionNos = new ArrayList<String>();
 		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);
 
@@ -302,17 +291,16 @@ private void generateDemandAndSendnotification(RequestInfo requestInfo, String t
 			CalculationReq calculationReq = CalculationReq.builder().calculationCriteria(calculationCriteriaList)
 					.requestInfo(requestInfo).isconnectionCalculation(true).isAdvanceCalculation(false).build();
 
-			Set<String> consumerCodes = new LinkedHashSet<String>();
-			consumerCodes.add(connectionNo);
 
-			if (!waterCalculatorDao.isDemandExists(tenantId, previousFromDate.getTimeInMillis(),
+
+			/*if (!waterCalculatorDao.isDemandExists(tenantId, previousFromDate.getTimeInMillis(),
 					previousToDate.getTimeInMillis(), consumerCodes)
 					&& !waterCalculatorDao.isConnectionExists(tenantId, previousFromDate.getTimeInMillis(),
 							previousToDate.getTimeInMillis(), consumerCodes)) {
 				log.warn("this connection doen't have the demand in previous billing cycle :" + connectionNo);
 				failedConnectionNos.add(connectionNo);
 				continue;
-			}
+			}*/
 			HashMap<Object, Object> genarateDemandData = new HashMap<Object, Object>();
 			genarateDemandData.put("calculationReq", calculationReq);
 			genarateDemandData.put("billingCycle",billingCycle);
@@ -577,9 +565,15 @@ 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() );
+		Long starTime = System.currentTimeMillis();
 		generateDemandAndSendnotification(bulkDemand.getRequestInfo(), bulkDemand.getTenantId(), billingPeriod, billingMasterData,
 				isSendMessage, isManual);
+		long endTime=System.currentTimeMillis();
+		long diff = endTime-starTime;
+		log.info("time takenn to generate demand for Tenantid:"+bulkDemand.getTenantId()+" BillPeriod:"+billingPeriod+" is : "+diff/1000 +" secs");
+
 
 	}
 	@KafkaListener(topics = {
@@ -592,6 +586,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 ;
@@ -604,8 +600,43 @@ public void generateDemandInBulkListner(HashMap<Object, Object> messageData) {
 		isSendMessage= (boolean) genarateDemandData.get("isSendMessage");
 		tenantId=(String) genarateDemandData.get("tenantId");
 
+		DateTimeFormatter formatter = DateTimeFormatter.ofPattern("d/MM/yyyy");
+
+
+		LocalDate fromDate = LocalDate.parse(billingCycle.split("-")[0].trim(), formatter);
+		LocalDate toDate = LocalDate.parse(billingCycle.split("-")[1].trim(), formatter);
+
+		Long dayStartTime = LocalDateTime
+				.of(fromDate.getYear(), fromDate.getMonth(), fromDate.getDayOfMonth(), 0, 0, 0)
+				.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+		Long dayEndTime = LocalDateTime
+				.of(toDate.getYear(), toDate.getMonth(), toDate.getDayOfMonth(), 23, 59, 59, 999000000)
+				.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+		Calendar previousFromDate = Calendar.getInstance();
+		Calendar previousToDate = Calendar.getInstance();
+
+		previousFromDate.setTimeInMillis(dayStartTime);
+		previousToDate.setTimeInMillis(dayEndTime);
+
+		previousFromDate.add(Calendar.MONTH, -1); //assuming billing cycle will be first day of month
+		previousToDate.add(Calendar.MONTH, -1);
+		int max = previousToDate.getActualMaximum(Calendar.DAY_OF_MONTH);
+		previousToDate.set(Calendar.DAY_OF_MONTH, max);
 		log.info("got generate demand call for :"+calculationReq.getCalculationCriteria().get(0).getConnectionNo());
-		generateDemandInBulk(calculationReq,billingCycle,masterMap,isSendMessage,tenantId);
+		Set<String> consumerCodes = new LinkedHashSet<String>();
+		consumerCodes.add(calculationReq.getCalculationCriteria().get(0).getConnectionNo());
+		if (!waterCalculatorDao.isDemandExists(tenantId, previousFromDate.getTimeInMillis(),
+				previousToDate.getTimeInMillis(), consumerCodes)
+				&& !waterCalculatorDao.isConnectionExists(tenantId, previousFromDate.getTimeInMillis(),
+				previousToDate.getTimeInMillis(), consumerCodes)) {
+			log.warn("this connection doen't have the demand in previous billing cycle :" + calculationReq.getCalculationCriteria().get(0).getConnectionNo());
+		} else {
+			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

municipal-services/ws-services/src/main/java/org/egov/waterconnection/repository/builder/WsQueryBuilder.java

@@ -67,6 +67,24 @@ public class WsQueryBuilder {
 			+ "eg_ws_connectionholder connectionholder ON connectionholder.connectionid = conn.id"
 			+ LEFT_OUTER_JOIN_STRING + "eg_ws_roadcuttinginfo roadcuttingInfo ON roadcuttingInfo.wsid = conn.id";
 
+	private static final String WATER_PLANE_SEARCH_QUERY = "SELECT count(*) OVER() AS full_count, conn.*, wc.*, document.*, plumber.*, wc.connectionCategory, wc.connectionType, wc.waterSource,"
+			+ " wc.meterId, wc.meterInstallationDate, wc.pipeSize, wc.noOfTaps, wc.proposedPipeSize, wc.proposedTaps, wc.connection_id as connection_Id, wc.connectionExecutionDate, wc.initialmeterreading, wc.appCreatedDate,"
+			+ " wc.detailsprovidedby, wc.estimationfileStoreId , wc.sanctionfileStoreId , wc.estimationLetterDate,"
+			+ " conn.id as conn_id, conn.tenantid, conn.applicationNo, conn.applicationStatus, conn.status, conn.connectionNo, conn.oldConnectionNo, conn.property_id, conn.roadcuttingarea,"
+			+ " conn.action, conn.adhocpenalty, conn.adhocrebate, conn.adhocpenaltyreason, conn.applicationType, conn.dateEffectiveFrom,"
+			+ " conn.adhocpenaltycomment, conn.adhocrebatereason, conn.adhocrebatecomment, conn.createdBy as ws_createdBy, conn.lastModifiedBy as ws_lastModifiedBy,"
+			+ " conn.createdTime as ws_createdTime, conn.lastModifiedTime as ws_lastModifiedTime,conn.additionaldetails, "
+			+ " conn.locality, conn.isoldapplication, conn.roadtype, document.id as doc_Id, document.documenttype, document.filestoreid, document.active as doc_active, plumber.id as plumber_id,"
+			+ " plumber.name as plumber_name, plumber.licenseno, roadcuttingInfo.id as roadcutting_id, roadcuttingInfo.roadtype as roadcutting_roadtype, roadcuttingInfo.roadcuttingarea as roadcutting_roadcuttingarea, roadcuttingInfo.roadcuttingarea as roadcutting_roadcuttingarea,"
+			+ " roadcuttingInfo.active as roadcutting_active, plumber.mobilenumber as plumber_mobileNumber, plumber.gender as plumber_gender, plumber.fatherorhusbandname, plumber.correspondenceaddress,"
+			+ " plumber.relationship, " + " {holderSelectValues}, " + " COALESCE(0) AS pendingamount, " + " COALESCE(0) AS taxamount, " + "{lastDemandDate}"
+			+ " FROM eg_ws_connection conn " + INNER_JOIN_STRING + " eg_ws_service wc ON wc.connection_id = conn.id"
+			+ LEFT_OUTER_JOIN_STRING + "eg_ws_applicationdocument document ON document.wsid = conn.id"
+			+ LEFT_OUTER_JOIN_STRING + "eg_ws_plumberinfo plumber ON plumber.wsid = conn.id" + LEFT_OUTER_JOIN_STRING
+			+ "eg_ws_connectionholder connectionholder ON connectionholder.connectionid = conn.id"
+			+ LEFT_OUTER_JOIN_STRING + "eg_ws_roadcuttinginfo roadcuttingInfo ON roadcuttingInfo.wsid = conn.id";
+
+
 	private static final String WATER_CONNNECTION_BY_DEMANNDDATE = "SELECT ((select distinct d.taxperiodto as taxperiodto from egbs_demand_v1 d inner join egbs_demanddetail_v1 dd on dd.demandid = d.id ";
 	private static final String WATER_CONNNECTION_BY_DEMANNDDATE_CLAUSE = " and  dd.taxheadcode='10101' and d.status = 'ACTIVE' and d.businessservice = 'WS' and d.consumercode = conn.connectionno order by d.taxperiodto desc limit 1))" +
 			" as taxperiodto, count(*) as count ";
@@ -662,7 +680,7 @@ public String getSearchQueryStringForPlaneSearch(SearchCriteria criteria, List<O
 			RequestInfo requestInfo) {
 		if (criteria.isEmpty())
 			return null;
-		StringBuilder query = new StringBuilder(WATER_SEARCH_QUERY);
+		StringBuilder query = new StringBuilder(WATER_PLANE_SEARCH_QUERY);
 		query = applyFiltersForPlaneSearch(query, preparedStatement, criteria);
 		return addPaginationWrapperForPlaneSearch(query.toString(), preparedStatement, criteria);
 	}

utilities/rollout-dashboard-cronjob-indexer/app.py

@@ -19,6 +19,7 @@ def getGPWSCHeirarchy():
     try:
         mdms_url = os.getenv('API_URL')
         state_tenantid = os.getenv('TENANT_ID')
+        url=os.getenv('IFIX_DEP_ENTITY_URL')
         mdms_requestData = {
             "RequestInfo": {
                 "apiId": "mgramseva-common",
@@ -56,7 +57,7 @@ def getGPWSCHeirarchy():
             if tenant.get('city') is not None and tenant.get('city').get('code') is not None:
                 teanant_data_Map.update({tenant.get('city').get('code'): tenant.get('code')})
 
-        url = 'https://mgramseva-dwss.punjab.gov.in/'
+        # url = 'https://mgramseva-dwss.punjab.gov.in/'
         print(url)
         requestData = {
             "requestHeader": {
@@ -926,32 +927,6 @@ def createEntryForRolloutToElasticSearch(tenant, activeUsersCount, totalAdvance,
 def process():
     print("continue is the process")
 
-    try:
-        connection = getConnection()
-        cursor = connection.cursor()
-
-        print("cursor: ", cursor)
-
-        DROPPING_TABLE_QUERY = " drop table if exists roll_out_dashboard "
-        cursor.execute(DROPPING_TABLE_QUERY)
-
-        connection.commit()
-
-        createTableQuery = createTable()
-        cursor.execute(createTableQuery)
-
-        connection.commit()
-
-        print("table dropped")
-    except Exception as exception:
-        print("Exception occurred while connecting to the database")
-        print(exception)
-
-    finally:
-        if connection:
-            cursor.close()
-            connection.close()
-
     tenants = getGPWSCHeirarchy()
     for tenant in tenants:
         print("Tenant:", tenant['tenantId'])
@@ -1071,4 +1046,4 @@ def createTable():
 
 
 if __name__ == '__main__':
-    process()
+    process()

utilities/rollout-dashboard-cronjob/app.py

@@ -17,6 +17,7 @@ def getGPWSCHeirarchy():
     try:
         mdms_url = os.getenv('API_URL')
         state_tenantid = os.getenv('TENANT_ID')
+        url=os.getenv('IFIX_DEP_ENTITY_URL')
         mdms_requestData = {
             "RequestInfo": {
                 "apiId": "mgramseva-common",
@@ -54,7 +55,7 @@ def getGPWSCHeirarchy():
             if tenant.get('city') is not None and tenant.get('city').get('code') is not None:
                 teanant_data_Map.update({tenant.get('city').get('code'): tenant.get('code')})
 
-        url = 'https://mgramseva-dwss.punjab.gov.in/'
+
         print(url)
         requestData = {
             "requestHeader": {
@@ -106,7 +107,7 @@ def getGPWSCHeirarchy():
                                         print(formatedTenantId)
                                         obj1 = {"tenantId": formatedTenantId, "zone": zone, "circle": circle,
                                                 "division": division, "subdivision": subdivision,
-                                                "section": section, "projectcode": tenantCode}
+                                                "section": section, "projectcode": tenantCode, "tenantName":tenantName}
                                         dataList.append(obj1)
         print("heirarchy collected")
         return (dataList)
@@ -792,8 +793,8 @@ def createEntryForRollout(tenant,activeUsersCount,totalAdvance, totalPenalty,tot
         createdTime = datetime.now(tz=tzInfo)
         print("createdtime -->", createdTime)
 
-        postgres_insert_query = "INSERT INTO roll_out_dashboard (tenantid, projectcode, zone, circle, division, subdivision, section,active_users_count,total_advance,total_penalty,total_connections,active_connections, last_demand_gen_date, demand_generated_consumer_count,total_demand_amount,collection_till_date,last_collection_date,expense_count,count_of_electricity_expense_bills,no_of_paid_expense_bills,last_expense_txn_date,total_amount_of_expense_bills,total_amount_of_electricity_bills,total_amount_of_paid_expense_bills,date_range,createdtime) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
-        record_to_insert = (tenant['tenantId'], tenant['projectcode'], tenant['zone'], tenant['circle'], tenant['division'], tenant['subdivision'], tenant['section'],activeUsersCount,totalAdvance, totalPenalty,totalConsumerCount,consumerCount,lastDemandGenratedDate,noOfDemandRaised,totaldemAmount,collectionsMade,lastCollectionDate, expenseCount,countOfElectricityExpenseBills,noOfPaidExpenseBills, lastExpTrnsDate, totalAmountOfExpenseBills, totalAmountOfElectricityBills, totalAmountOfPaidExpenseBills,date, createdTime)
+        postgres_insert_query = "INSERT INTO roll_out_dashboard (tenantid, tenantName, projectcode, zone, circle, division, subdivision, section,active_users_count,total_advance,total_penalty,total_connections,active_connections, last_demand_gen_date, demand_generated_consumer_count,total_demand_amount,collection_till_date,last_collection_date,expense_count,count_of_electricity_expense_bills,no_of_paid_expense_bills,last_expense_txn_date,total_amount_of_expense_bills,total_amount_of_electricity_bills,total_amount_of_paid_expense_bills,date_range,createdtime) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"
+        record_to_insert = (tenant['tenantId'],tenant['tenantName'], tenant['projectcode'], tenant['zone'], tenant['circle'], tenant['division'], tenant['subdivision'], tenant['section'],activeUsersCount,totalAdvance, totalPenalty,totalConsumerCount,consumerCount,lastDemandGenratedDate,noOfDemandRaised,totaldemAmount,collectionsMade,lastCollectionDate, expenseCount,countOfElectricityExpenseBills,noOfPaidExpenseBills, lastExpTrnsDate, totalAmountOfExpenseBills, totalAmountOfElectricityBills, totalAmountOfPaidExpenseBills,date, createdTime)
         cursor.execute(postgres_insert_query, record_to_insert)
 
         connection.commit()
@@ -893,6 +894,7 @@ def createTable():
     CREATE_TABLE_QUERY = """create table roll_out_dashboard(
         id SERIAL primary key, 	
         tenantid varchar(250) NOT NULL,
+        tenantName varchar(500),
         projectcode varchar(66),
         zone varchar(250),
         circle varchar(250),