Skip to content
This repository has been archived by the owner on Nov 26, 2024. It is now read-only.

Added failed connection logic in generate demand Bulk topic #884

Open
wants to merge 12 commits into
base: master
Choose a base branch
from
Open

Added failed connection logic in generate demand Bulk topic #884

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
4621328
Added failed connection logic in generate demand Bulk topic
Aug 7, 2024
e356e3d
Added failed connection logic in generate demand Bulk topic
Aug 7, 2024
6402055
Added Log Linne for All the Time taken in Calls
Aug 8, 2024
5a1b561
Added Log Linne for All the Time taken in Calls
Aug 8, 2024
9cf61b5
Added Log Linne for All the Time taken in Calls
Aug 8, 2024
e3decb9
Added kafka properties
Aug 8, 2024
0c673b8
Added kafka properties
Aug 8, 2024
75277c3
Added kafka properties
Aug 8, 2024
10f1312
Added kafka properties
Aug 8, 2024
6be5adf
Removed ack From properties
Aug 8, 2024
b9fe4c4
ADDED spring kafka configs
Aug 8, 2024
3ffcc52
ADDED spring kafka configs
Aug 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 68 additions & 16 deletions ...ws-calculator/src/main/java/org/egov/wscalculation/consumer/DemandGenerationConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.egov.wscalculation.web.models.users.UserDetailResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
Expand Down Expand Up @@ -71,6 +72,9 @@ public class DemandGenerationConsumer {
@Autowired
private NotificationUtil util;

@Autowired
private KafkaTemplate kafkaTemplate;

@Autowired
private CalculatorUtil calculatorUtils;

Expand Down Expand Up @@ -192,7 +196,7 @@ private void generateDemandInBatch(CalculationReq request, Map<String, Object> m
wsCalulationWorkflowValidator.applicationValidation(request.getRequestInfo(), criteria.getTenantId(),
criteria.getConnectionNo(), genratedemand);
}*/
System.out.println("Calling Bulk Demand generation connection Number" + request.getCalculationCriteria().get(0).getConnectionNo());
//System.out.println("Calling Bulk Demand generation connection Number" + request.getCalculationCriteria().get(0).getConnectionNo());
wSCalculationServiceImpl.bulkDemandGeneration(request, masterMap);
/*String connectionNoStrings = request.getCalculationCriteria().stream()
.map(criteria -> criteria.getConnectionNo()).collect(Collectors.toSet()).toString();
Expand Down Expand Up @@ -269,12 +273,13 @@ private void generateDemandAndSendnotification(RequestInfo requestInfo, String t
.of(toDate.getYear(), toDate.getMonth(), toDate.getDayOfMonth(), 23, 59, 59, 999000000)
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();


Long StartTimeForGetConnetion = System.currentTimeMillis();
List<String> connectionNos = waterCalculatorDao.getNonMeterConnectionsList(tenantId, dayStartTime, dayEndTime);
log.info("Time taken for the for Get Connetions : "+(System.currentTimeMillis()-StartTimeForGetConnetion)/1000+ " Secondss");



Calendar previousFromDate = Calendar.getInstance();
/* Calendar previousFromDate = Calendar.getInstance();
Calendar previousToDate = Calendar.getInstance();

previousFromDate.setTimeInMillis(dayStartTime);
Expand All @@ -283,26 +288,27 @@ private void generateDemandAndSendnotification(RequestInfo requestInfo, String t
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);
previousToDate.set(Calendar.DAY_OF_MONTH, max);*/
String assessmentYear = estimationService.getAssessmentYear();
ArrayList<String> failedConnectionNos = new ArrayList<String>();
Long startTimeForMdms= System.
currentTimeMillis();
Map<String, Object> masterMap = mDataService.loadMasterData(requestInfo,
tenantId);
log.info("Time taken for the for Get Mdms Data : "+(System.currentTimeMillis()-startTimeForMdms)/1000+ " Secondss");

log.info("connectionNos" + connectionNos.size());
log.info("connectionNos" + connectionNos);
log.info("dayStartTime:"+dayStartTime);
log.info("dayEndTime"+dayEndTime);

long startTimeForLoop= System.currentTimeMillis();
for (String connectionNo : connectionNos) {
long timeBeforePushToKafka = System.currentTimeMillis();
CalculationCriteria calculationCriteria = CalculationCriteria.builder().tenantId(tenantId)
.assessmentYear(assessmentYear).connectionNo(connectionNo).from(dayStartTime).to(dayEndTime).build();
List<CalculationCriteria> calculationCriteriaList = new ArrayList<>();
calculationCriteriaList.add(calculationCriteria);
CalculationReq calculationReq = CalculationReq.builder().calculationCriteria(calculationCriteriaList)
.requestInfo(requestInfo).isconnectionCalculation(true).isAdvanceCalculation(false).build();

Set<String> consumerCodes = new LinkedHashSet<String>();
/*Set<String> consumerCodes = new LinkedHashSet<String>();
consumerCodes.add(connectionNo);

if (!waterCalculatorDao.isDemandExists(tenantId, previousFromDate.getTimeInMillis(),
Expand All @@ -312,13 +318,14 @@ private void generateDemandAndSendnotification(RequestInfo requestInfo, String t
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);
genarateDemandData.put("masterMap",masterMap);
genarateDemandData.put("isSendMessage",isSendMessage);
genarateDemandData.put("tenantId",tenantId);
log.info("Time taken before push to kafka : "+(System.currentTimeMillis()-timeBeforePushToKafka)+ " in ms");

/*
* List<Demand> demands = demandService.searchDemand(tenantId, consumerCodes,
Expand All @@ -327,10 +334,15 @@ private void generateDemandAndSendnotification(RequestInfo requestInfo, String t
* log.warn("this connection doen't have the demand in previous billing cycle :"
* + connectionNo ); continue; }
*/
log.info("sending generate demand for connection no :"+connectionNo);
producer.push(config.getWsGenerateDemandBulktopic(),genarateDemandData);
//log.info("sending generate demand for connection no :"+connectionNo);
long timetakenToPush= System.currentTimeMillis();
kafkaTemplate.send(config.getWsGenerateDemandBulktopic(),genarateDemandData);
log.info("Time taken for the for push to kafka : "+(System.currentTimeMillis()-timetakenToPush)+ " in ms");

}
log.info("Time taken for the for loop : "+(System.currentTimeMillis()-startTimeForLoop)/1000+ " Secondss");

Long starttimeforNotification= System.currentTimeMillis();
HashMap<String, String> demandMessage = util.getLocalizationMessage(requestInfo,
WSCalculationConstant.mGram_Consumer_NewDemand, tenantId);
HashMap<String, String> gpwscMap = util.getLocalizationMessage(requestInfo, tenantId, tenantId);
Expand Down Expand Up @@ -364,6 +376,7 @@ private void generateDemandAndSendnotification(RequestInfo requestInfo, String t
}

});
log.info("Time taken for notification : "+(System.currentTimeMillis()-starttimeforNotification)/1000+ " Secondss");
/* if (isSendMessage && failedConnectionNos.size() > 0) {
List<ActionItem> actionItems = new ArrayList<>();
String actionLink = config.getBulkDemandFailedLink();
Expand Down Expand Up @@ -577,10 +590,14 @@ 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 +" seconds");
}
@KafkaListener(topics = {
"${egov.update.demand.add.penalty}" })
Expand All @@ -603,9 +620,44 @@ public void generateDemandInBulkListner(HashMap<Object, Object> messageData) {
billingCycle= (String) genarateDemandData.get("billingCycle");
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);

log.info("got generate demand call for :"+calculationReq.getCalculationCriteria().get(0).getConnectionNo());
generateDemandInBulk(calculationReq,billingCycle,masterMap,isSendMessage,tenantId);
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);
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Refactor Calendar Usage to LocalDate.

Replace Calendar with LocalDate for consistency and to leverage the modern date-time API.

LocalDate previousFromDate = fromDate.minusMonths(1);
LocalDate previousToDate = toDate.minusMonths(1).with(TemporalAdjusters.lastDayOfMonth());

//log.info("got generate demand call for :"+calculationReq.getCalculationCriteria().get(0).getConnectionNo());
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 {
Long starttime = System.currentTimeMillis();
generateDemandInBulk(calculationReq, billingCycle, masterMap, isSendMessage, tenantId);
log.info("GOt call inn ws-gennerate-demand-bulk topic end time:" + System.currentTimeMillis());
Long endtime = System.currentTimeMillis();
long diff = endtime - starttime;
log.info("Time taken to process request for :" + calculationReq.getCalculationCriteria().get(0).getConnectionNo() + " is :" + diff / 1000 + " secs");
}
}

}
2 changes: 2 additions & 0 deletions ...s/ws-calculator/src/main/java/org/egov/wscalculation/repository/WSCalculationDaoImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,8 @@ public List<String> getConnectionsNoList(String tenantId, String connectionType)
public List<String> getNonMeterConnectionsList(String tenantId, Long dayStartTime, Long dayEndTime) {
List<Object> preparedStatement = new ArrayList<>();
String query = queryBuilder.getNonMeteredConnectionsList(tenantId, dayStartTime, dayEndTime, preparedStatement);
log.info("Query: "+query );
log.info("Prepare Statement : "+preparedStatement.toArray() );
return jdbcTemplate.query(query, preparedStatement.toArray(), demandSchedulerRowMapper);
}

Expand Down
2 changes: 1 addition & 1 deletion .../ws-calculator/src/main/java/org/egov/wscalculation/service/WSCalculationServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -295,7 +295,7 @@ List<Calculation> getCalculations(CalculationReq request, Map<String, Object> ma
List<Calculation> calculations = new ArrayList<>(request.getCalculationCriteria().size());
for (CalculationCriteria criteria : request.getCalculationCriteria()) {
Map<String, List> estimationMap = null;
log.info("Innside get Calculationn");
//.info("Innside get Calculationn");
if(request.getIsAdvanceCalculation() == null || (!request.getIsAdvanceCalculation().booleanValue())) {
estimationMap = estimationService.getEstimationMap(criteria, request.getRequestInfo(),
masterMap,request.getIsconnectionCalculation(),false);
Expand Down
7 changes: 6 additions & 1 deletion municipal-services/ws-calculator/src/main/resources/application.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,10 @@ spring.kafka.consumer.group-id=egov-ws-calc-services
spring.kafka.consumer.properties.spring.json.use.type.headers=false
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.acks=0
spring.kafka.producer.linger.ms=10
spring.kafka.producer.batch.size=32768
spring.kafka.producer.buffer.memory=33554432

$KAFKA TOPIC DETAILS
egov.watercalculatorservice.createdemand.topic=ws-generate-demand
Expand Down Expand Up @@ -158,4 +162,5 @@ 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