class MarriageRegnConsumer {

	@KafkaListener(topics = { "${kafka.topics.create.fee}", "${kafka.topics.update.fee}",
			"${kafka.topics.create.reissueCertificate}", "${kafka.topics.update.reissueappl}",
			"${kafka.topics.create.reissueappl}", "${kafka.topics.create.registrationunit}",
			"${kafka.topics.update.registrationunit}", 
			 "${kafka.topics.create.marriagedocumenttype}",
			"${kafka.topics.update.marriagedocumenttype}" })
	public void processMessage(@Payload Map<String, Object> consumerRecord,
			@Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
		log.info("Entered kakfaConsumer "+consumerRecord);

		if (topic.equals(propertiesManager.getCreateMarriageRegnTopicName())) {
			log.info("entering create marriageRegn consumer");
			marriageRegnService.create(objectMapper.convertValue(consumerRecord, MarriageRegnRequest.class));
		}
		if (topic.equals(propertiesManager.getUpdateMarriageRegnTopicName())) {
			log.info("entering update marriageRegn consumer");
			marriageRegnService.update(objectMapper.convertValue(consumerRecord, MarriageRegnRequest.class));
		}

		if (topic.equals(propertiesManager.getCreateRegistrationUnitTopicName())) {
			log.info("entering update_RegistrationUnit consumer");
			registrationUnitService.create(objectMapper.convertValue(consumerRecord, RegnUnitRequest.class));
		}
		if (topic.equals(propertiesManager.getUpdateRegistrationUnitTopicName())) {
			log.info("entering update_RegistrationUnit consumer");
			registrationUnitService.update(objectMapper.convertValue(consumerRecord, RegnUnitRequest.class));
		}

		if (topic.equals(propertiesManager.getCreateMarriageDocumentTypeTopicName())) {
			marriageDocumentTypeService.create(objectMapper.convertValue(consumerRecord, MarriageDocTypeRequest.class));
		}
		if (topic.equals(propertiesManager.getUpdateMarriageDocumentTypeTopicName())) {
			marriageDocumentTypeService.update(objectMapper.convertValue(consumerRecord, MarriageDocTypeRequest.class));
		}
		if (topic.equals(propertiesManager.getCreateReissueMarriageRegnTopicName()))
			marriageCertService.create(objectMapper.convertValue(consumerRecord, ReissueCertRequest.class));
		if (topic.equals(propertiesManager.getUpdateReissueMarriageRegnTopicName()))
			marriageCertService.update(objectMapper.convertValue(consumerRecord, ReissueCertRequest.class));
		if (topic.equals(propertiesManager.getCreateReissueCertificateTopicName()))
			marriageCertService.createCert(objectMapper.convertValue(consumerRecord, ReissueCertAppl.class));
		if (topic.equals(propertiesManager.getCreateFeeTopicName()))
			feeService.createFee(objectMapper.convertValue(consumerRecord, FeeRequest.class));
		if (topic.equals(propertiesManager.getUpdateFeeTopicName()))
			feeService.updateFee(objectMapper.convertValue(consumerRecord, FeeRequest.class));
	}

}