package com.mns.oms.batch.processor;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import com.mns.oms.batch.domain.OMSMetaDataLive;
import com.mns.oms.batch.exception.DataNotFoundException;
import com.mns.oms.batch.proxy.DncOrderServiceProxy;
import com.mns.oms.batch.proxy.ReturnsAPIServiceProxy;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

import com.mns.oms.batch.constants.BatchConstant;
import com.mns.oms.batch.domain.CarrierData;
import com.mns.oms.batch.mapper.BeamDataMapper;
import com.mns.oms.batch.mapper.CycleAvoidingMappingContext;
import com.mns.oms.batch.model.BeamDataDTO;
import com.mns.oms.batch.service.CarrierDataService;
import org.springframework.http.ResponseEntity;

public class BeamDataProcessor implements ItemProcessor<CarrierData, BeamDataDTO> {
	private static final Logger log = LoggerFactory.getLogger(BeamDataProcessor.class);

	@Autowired
	@Qualifier("batchconfigs")
	private Map<String, String> batchAppConfigs;

	@Autowired
	private CarrierDataService carrierDataService;


	@Autowired
	private DncOrderServiceProxy orderDetailProxy;

	@Autowired
	private ReturnsAPIServiceProxy returnsAPIServiceProxy;

	@Override
	public BeamDataDTO process(CarrierData item) throws Exception {

		log.debug("Processing carrier data for BEAM----->{}", item);

		String refundPromiseDate;
		String extCustomerId = null;
		String id = item.getId();
		String statusCode = item.getStatusDetails().getStatusCode();

		String messageID = "BEAM_WISMR_" + id + "_"
				+ (LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss")));

		
			if (StringUtils.isEmpty(item.getOrderRef()) || StringUtils.isEmpty(item.getReturnOrderNo())) {
				log.debug("Getting the orderReferenceNumber and returnOrderNum for status code: {}, Item parcelCode: {}, Item carrierCode: {} ",
						BatchConstant.FIRST_STATUS_CODE,
						item.getParcelCode(),
						item.getCarrierCode());
				// Retrieving the CarrierData object for the item's parcel code, status code, and carrier code
				CarrierData carrierData = getCarrierDataForParcelCodeStatusCode(item.getParcelCode(), BatchConstant.FIRST_STATUS_CODE, item.getCarrierCode());
				// Retrieving the external customer ID from the CarrierData object
				extCustomerId = (carrierData != null) ? getExtCustomerId(carrierData) : null;
			}
		
		//When both the getOrderRef and getReturnOrderNo values are available,
		// the extCustomerId is obtained by searching for a matching orderref in
		// either the oms-metadata-live collection or the OMS Oracle database.
		if(StringUtils.isEmpty(extCustomerId)){
			extCustomerId =getExtCustomerId(item);
		}
		log.info("Retrieved extCustomerId: {}", extCustomerId);
		item.setExtCustomerId(extCustomerId);
		item.setMessageID(messageID);
		
		log.debug("Carrier data before converting to DTO----->{}", item);
		
		BeamDataDTO beamDataDTO = BeamDataMapper.INSTANCE.getBeamDetails(item, new CycleAvoidingMappingContext());
		
		log.debug("Beam data DTO being sent to writer----->{}", beamDataDTO);

		return beamDataDTO;

	}


	public String getExtCustomerId(CarrierData carrierData) {
//		carrierData.setOrderRef("301-0000258-5637798");
//		carrierData.setOrderRef("301-20230713-01362");
//		carrierData.setReturnOrderNo("0000");
		if (StringUtils.isNotEmpty(carrierData.getOrderRef()) && StringUtils.isNotEmpty(carrierData.getReturnOrderNo())) {
			log.debug("Getting the extCustomerId against the order reference and return order number: {} - {}", carrierData.getOrderRef(), carrierData.getReturnOrderNo());
			// Query OMS Live collection to get extCustomerId and Get the external customer ID from the OMS live collection
			OMSMetaDataLive omsMetaDataLive = getOMSLiveCollection(carrierData.getOrderRef(), carrierData.getReturnOrderNo());
			if (omsMetaDataLive != null) {
				log.info("After omsMetaDataLive is found in getOMSLiveCollection, getting the extCustomerId : {}", omsMetaDataLive.getExternalCustomerId());
				return omsMetaDataLive.getExternalCustomerId();
			} else {
				log.info("omsMetaDataLive is not found in getOMSLiveCollection, querying the OMS Oracle database for the extCustomerId");
				// If not found, query the OMS Oracle database
				ResponseEntity<Object> omsDBData = queryOMSDB(carrierData.getOrderRef());
				LinkedHashMap<String, Object> omsMetaDataLiveMap = (LinkedHashMap<String, Object>) omsDBData.getBody();
				log.info("After querying the OMS Oracle database for the extCustomerId: {}",String.valueOf(omsMetaDataLiveMap.get("externalCustomerId")));
				return String.valueOf(omsMetaDataLiveMap.get("externalCustomerId"));
			}
		}else{
			throw new DataNotFoundException("No data found for order no: " +carrierData.getOrderRef());
		}
	}

	private OMSMetaDataLive getOMSLiveCollection(String orderRef, String returnOrderNo) {
		log.info("Entering getOMSLiveCollection method with orderRef: {}", orderRef);
		// Query OMS Live using mongo repository collection using feign to get extCustomerId
		ResponseEntity<List<OMSMetaDataLive>> omsMetaDataLiveResponse = returnsAPIServiceProxy.searchOrderWithOrderNo(orderRef);
		log.info("Received OMSMetaDataLiveResponse for order ref {}: {}", orderRef, omsMetaDataLiveResponse);
		// Extract the first element from the response list or else null
		OMSMetaDataLive omsMetaDataLive = omsMetaDataLiveResponse.getBody().stream()
				.filter(omsMetaDataLiveFilter -> {
					// Filter out any OMSMetaDataLive objects that don't have a non-empty ID field
					return !StringUtils.isEmpty(omsMetaDataLiveFilter.getExternalCustomerId());
				})
				.findFirst()
				.orElse(null);
		log.info("Received omsMetaDataLive in response body from searchOrderWithOrderNo: {}", omsMetaDataLive);
		// Return the extracted OMSMetaDataLive object
		return omsMetaDataLive;
	}

	public ResponseEntity<Object> fetchOrderDetailsFromOrderService(String orderNumber){
		log.info("Calling DNC order service API by feign to get the order details for an orderNumber: {}", orderNumber);
		String orderInput = "{\"type\":\"OrderNo\",\"value\":\"" + orderNumber + "\"}";
		return orderDetailProxy.getOrderDetails(orderInput);
	}
	private ResponseEntity<Object> queryOMSDB(String orderNumber) {
		// the code queries the OMS Oracle database using a Feign client in the Beam Batch service.
		//orderdetailsforreturnbyid(orderRef)
		ResponseEntity<Object> responseEntity = fetchOrderDetailsFromOrderService(orderNumber);
		log.info("Response fetched after calling DNC order service API by feign to get the order details for an orderNumber: {} for response: {}",orderNumber, responseEntity.getBody());
		return responseEntity;
	}
	public CarrierData getCarrierDataForParcelCodeStatusCode(String parcelCode, String statusCode, String carrierCode) {
		return carrierDataService.getCarrierDataForParcelCodeStatusCode(parcelCode, statusCode,
				carrierCode);
	}


}


package com.mns.oms.batch.writer;

import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mns.oms.batch.domain.CarrierData;
import com.mns.oms.batch.model.BeamDataDTO;
import com.mongodb.client.result.UpdateResult;

public class KafkaBatchWriter implements ItemWriter<BeamDataDTO> {

	private static final Logger log = LoggerFactory.getLogger(KafkaBatchWriter.class);

	@Value("${kafka.topic.omsapi.beam}")
	private String topic;

	@Autowired
	MongoTemplate mongoTemplate;

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	@Override
	public void write(List<? extends BeamDataDTO> items) throws Exception {

		List<String> idList = new ArrayList<>();
		items.forEach(item -> {

			try {

				ObjectMapper objectMapper = new ObjectMapper();
				String data = objectMapper.writeValueAsString(item);
				log.debug("Data to be sent to BEAM------->{}", data);
				
//				kafkaTemplate.send(topic, data);
				
				idList.add(item.getMessageID().split("_")[2]);
			} catch (JsonProcessingException e) {
				// TODO Auto-generated catch block
				log.error("ERROR in SENDING DATA to BEAM---------------{}", e);
			}

		});

		// Batch update
		if (idList.size() > 0) {
			Query recordsToUpdateQuery = Query.query(Criteria.where("_id").in(idList));
			UpdateResult updtResult = mongoTemplate.updateMulti(recordsToUpdateQuery,
					Update.update("isProcessed", true), CarrierData.class);
			log.info("Rows matched to update records in DB after sending data to BEAM------------{}",
					updtResult.getMatchedCount());
			log.info("Rows modified in DB after sending data to BEAM------------{}", updtResult.getModifiedCount());
		}
	}
}

In presnet behavoiur.....throw new DataNotFoundException("No data found for order no: " +carrierData.getOrderRef());....make sure that code
UpdateResult updtResult = mongoTemplate.updateMulti(recordsToUpdateQuery,Update.update("isProcessed", true), CarrierData.class); is not executed after
after throwing of exception ....how to achieve this thing in another way without throwing exception that .....isProcessed true is not updated....if same 
condition if (StringUtils.isNotEmpty(carrierData.getOrderRef()) && StringUtils.isNotEmpty(carrierData.getReturnOrderNo())) { else block