Merge pull request #110 from sunbird-cb/dup_records_fixes

Dup records fixes
sunbird-cb · Jul 20, 2022 · 66ae2f5 · 66ae2f5
2 parents f839b84 + 6f7ad89
commit 66ae2f5
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 18 deletions.
diff --git a/src/main/java/org/sunbird/common/util/CbExtServerProperties.java b/src/main/java/org/sunbird/common/util/CbExtServerProperties.java
@@ -1,11 +1,11 @@
 package org.sunbird.common.util;
 
-import java.util.Arrays;
-import java.util.List;
-
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
+import java.util.Arrays;
+import java.util.List;
+
 @Component
 public class CbExtServerProperties {
 
@@ -308,16 +308,16 @@ public class CbExtServerProperties {
 
 	@Value("${user.registration.custodian.orgName}")
 	private String custodianOrgName;
-	
+
 	@Value("${user.position.master.list.file}")
 	private String masterPositionListFileName;
-	
+
 	@Value("${user.registration.welcome.email.template}")
 	private String welcomeEmailTemplate;
-	
+
 	@Value("${user.registration.welcome.email.subject}")
 	private String welcomeEmailSubject;
-	
+
 	@Value("${sb.org.create.path}")
 	private String lmsOrgCreatePath;
 

diff --git a/src/main/java/org/sunbird/user/registration/service/UserRegistrationConsumer.java b/src/main/java/org/sunbird/user/registration/service/UserRegistrationConsumer.java
@@ -54,9 +54,7 @@ public class UserRegistrationConsumer {
 	UserRegistrationNotificationService userRegNotificationService;
 
 	@SuppressWarnings("unchecked")
-	@KafkaListener(topicPartitions = {
-			@TopicPartition(topic = "${kafka.topics.user.registration.register.event}", partitions = { "0", "1", "2",
-					"3" }) })
+	@KafkaListener(topics = "${kafka.topics.user.registration.register.event}",groupId = "${kafka.topics.user.registration.register.event.consumer.group}" )
 	public void processMessage(ConsumerRecord<String, String> data) {
 		UserRegistration userRegistration = gson.fromJson(data.value(), UserRegistration.class);
 		/*
@@ -98,9 +96,7 @@ public void processMessage(ConsumerRecord<String, String> data) {
 		userRegNotificationService.sendNotification(userRegistration);
 	}
 
-	@KafkaListener(topicPartitions = {
-			@TopicPartition(topic = "${kafka.topics.user.registration.createUser}", partitions = { "0", "1", "2",
-					"3" }) })
+	@KafkaListener(topics = "${kafka.topics.user.registration.createUser}",groupId = "${kafka.topics.user.registration.createUser.consumer.group}")
 	public void processCreateUserMessage(ConsumerRecord<String, String> data) {
 		try {
 			WfRequest wfRequest = gson.fromJson(data.value(), WfRequest.class);
@@ -112,9 +108,7 @@ public void processCreateUserMessage(ConsumerRecord<String, String> data) {
 		}
 	}
 
-	@KafkaListener(topicPartitions = {
-			@TopicPartition(topic = "${kafka.topics.user.registration.auto.createUser}", partitions = { "0", "1", "2",
-					"3" }) })
+	@KafkaListener(topics = "${kafka.topics.user.registration.auto.createUser}",groupId = "${kafka.topics.user.registration.auto.createUser.consumer.group}")
 	public void processAutoCreateUserEvent(ConsumerRecord<String, String> data) {
 		try {
 			UserRegistration userRegistration = gson.fromJson(data.value(), UserRegistration.class);

diff --git a/src/main/java/org/sunbird/workallocation/service/WATConsumer.java b/src/main/java/org/sunbird/workallocation/service/WATConsumer.java
@@ -62,8 +62,7 @@ public class WATConsumer {
 	private static final String[] ignorableFieldsForPublishedState = { "userName", "userEmail", "submittedFromName",
 			"submittedFromEmail", "submittedToName", "submittedToEmail", "createdByName", "updatedByName" };
 
-	@KafkaListener(id = "id2", groupId = "watTelemetryTopic-consumer", topicPartitions = {
-			@TopicPartition(topic = "${kafka.topics.wat.telemetry.event}", partitions = { "0", "1", "2", "3" }) })
+	@KafkaListener(id = "id2", topics = "${kafka.topics.wat.telemetry.event}", groupId = "${kafka.topics.wat.telemetry.event.topic.consumer}")
 	public void processMessage(ConsumerRecord<String, String> data) {
 		try {
 			logger.info("Consuming the audit records for WAT .....");

diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
@@ -66,6 +66,7 @@ kafka.topics.wat.telemetry.event=watTelemetryTopic
 kafka.topics.parent.telemetry.event=dev.telemetry.raw
 kafka.topics.parent.rating.event = dev.rating.event
 kafka.topics.userutility.telemetry.event=userLastLoginTopic
+kafka.topics.wat.telemetry.event.topic.consumer=watTelemetryTopic-consumer
 
 userutility.telemetry.event.pdata.id=dev.sunbird.cb.ext.service
 userutility.telemetry.event.pdata.pid=sunbird-cb-ext-service
@@ -187,6 +188,9 @@ es.user.registration.index=user_registration
 kafka.topics.user.registration.register.event=user.register.event
 kafka.topics.user.registration.createUser=workflow.user.registration.createUser
 kafka.topics.user.registration.auto.createUser=user.register.createUser.event
+kafka.topics.user.registration.auto.createUser.consumer.group = userAutoRegistrationTopic-consumer
+kafka.topics.user.registration.createUser.consumer.group = userRegistrationTopic-consumer
+kafka.topics.user.registration.register.event.consumer.group = userRegistrationRegisterEventTopic-consumer
 #Provide Dept Id as comma separated values here to exclude the details
 user.registration.dept.exclude.list=0133334975707217922
 user.registration.workflow.service.name=user_registration