diff --git a/Dockerfile b/Dockerfile
index 43b9ddc..a3c7ef3 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,7 @@
 FROM sivaprakash123/openjdk:17-slim-debian11
+
+RUN useradd -ms /bin/bash appuser
+
 # Install necessary dependencies
 RUN apt-get update \
     && apt-get install -y \
@@ -12,6 +15,12 @@ RUN apt-get update \
         xz-utils \
     && apt-get clean \
     && rm -rf /var/lib/apt/lists/*
+    
 COPY cb-ext-userprofile-service-1.0-SNAPSHOT.jar /opt/
+
+RUN chown -R appuser:appuser /opt
+USER appuser
+WORKDIR /opt
+
 #HEALTHCHECK --interval=30s --timeout=30s CMD curl --fail http://localhost:7001/actuator/health || exit 1
 CMD ["/bin/bash", "-c", "java -XX:+PrintFlagsFinal $JAVA_OPTIONS -XX:+UnlockExperimentalVMOptions -jar /opt/cb-ext-userprofile-service-1.0-SNAPSHOT.jar"]
diff --git a/pom.xml b/pom.xml
index 7c06953..9f683ca 100644
--- a/pom.xml
+++ b/pom.xml
@@ -192,6 +192,11 @@
             <version>5.9.2</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.auth0</groupId>
+            <artifactId>java-jwt</artifactId>
+            <version>4.4.0</version>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>
diff --git a/src/main/java/com/igot/cb/profile/controller/AchievementController.java b/src/main/java/com/igot/cb/profile/controller/AchievementController.java
new file mode 100644
index 0000000..c0e8a5f
--- /dev/null
+++ b/src/main/java/com/igot/cb/profile/controller/AchievementController.java
@@ -0,0 +1,71 @@
+package com.igot.cb.profile.controller;
+
+import com.igot.cb.profile.service.AchievementService;
+import com.igot.cb.transactional.elasticsearch.dto.SearchCriteria;
+import com.igot.cb.util.ApiResponse;
+import com.igot.cb.util.Constants;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.Map;
+
+@RestController
+@RequestMapping("/learner/achievement")
+public class AchievementController {
+
+    @Autowired
+    private AchievementService achievementService;
+
+    @PostMapping("/create")
+    public ResponseEntity<?> createLearnerAchievement(
+            @RequestHeader(value = Constants.X_AUTH_TOKEN, required = true) String authToken,
+            @RequestHeader(value = Constants.X_AUTH_USER_ORG_ID, required = true) String rootOrgId,
+            @RequestBody Map<String, Object> request) throws Exception {
+        ApiResponse response = achievementService.createLearnerAchievement(request, authToken, rootOrgId);
+        return new ResponseEntity<>(response, response.getResponseCode());
+    }
+
+    @PutMapping("/update")
+    public ResponseEntity<?> updateLearnerAchievement(
+            @RequestHeader(value = Constants.X_AUTH_TOKEN, required = true) String authToken,
+            @RequestHeader(value = Constants.X_AUTH_USER_ORG_ID, required = true) String rootOrgId,
+            @RequestBody Map<String, Object> request) throws Exception {
+        ApiResponse response = achievementService.updateLearnerAchievement(request, authToken, rootOrgId);
+        return new ResponseEntity<>(response, response.getResponseCode());
+    }
+
+    @GetMapping("/read/{achievementId}")
+    public ResponseEntity<Object> readLearnerAchievement(@PathVariable(Constants.ACHIEVEMENT_ID) String achievementId,
+                                                         @RequestHeader(value = Constants.X_AUTH_TOKEN, required = true) String authToken,
+                                                         @RequestHeader(value = Constants.CONTEXT_TYPE, required = true) String contextType) {
+        ApiResponse response = achievementService.readLearnerAchievement(achievementId, authToken, contextType);
+        return new ResponseEntity<>(response, HttpStatus.valueOf(response.getResponseCode().value()));
+    }
+
+    @DeleteMapping("/delete")
+    public ResponseEntity<?> deleteLearnerAchievement(
+            @RequestHeader(value = Constants.X_AUTH_TOKEN, required = true) String authToken,
+            @RequestBody Map<String, Object> request) throws Exception {
+        ApiResponse response = achievementService.deleteLearnerAchievement(request, authToken);
+        return new ResponseEntity<>(response, response.getResponseCode());
+    }
+
+    @PutMapping("/status/update")
+    public ResponseEntity<?> statusUpdateLearnerAchievement(
+            @RequestHeader(value = Constants.X_AUTH_TOKEN, required = false) String authToken,
+            @RequestBody Map<String, Object> request) {
+        ApiResponse response = achievementService.statusUpdateLearnerAchievement(request, authToken);
+        return new ResponseEntity<>(response, response.getResponseCode());
+    }
+
+    @PostMapping("/search")
+    public ResponseEntity<?> searchLearnerAchievements(
+            @RequestHeader(value = Constants.X_AUTH_TOKEN, required = true) String authToken,
+            @RequestBody SearchCriteria searchCriteria) {
+        ApiResponse response = achievementService.searchLearnerAchievements(searchCriteria, authToken);
+        return new ResponseEntity<>(response, response.getResponseCode());
+    }
+
+}
diff --git a/src/main/java/com/igot/cb/profile/service/AchievementService.java b/src/main/java/com/igot/cb/profile/service/AchievementService.java
new file mode 100644
index 0000000..1c1d411
--- /dev/null
+++ b/src/main/java/com/igot/cb/profile/service/AchievementService.java
@@ -0,0 +1,21 @@
+package com.igot.cb.profile.service;
+
+import com.igot.cb.transactional.elasticsearch.dto.SearchCriteria;
+import com.igot.cb.util.ApiResponse;
+
+import java.util.Map;
+
+public interface AchievementService {
+
+    ApiResponse createLearnerAchievement(Map<String, Object> request, String userToken, String rootOrgId);
+
+    ApiResponse updateLearnerAchievement(Map<String, Object> request, String userToken, String orgToken);
+
+    ApiResponse deleteLearnerAchievement(Map<String, Object> request, String userToken);
+
+    ApiResponse readLearnerAchievement(String achievementId, String userToken, String contextType);
+
+    ApiResponse statusUpdateLearnerAchievement(Map<String, Object> request, String authToken);
+
+    ApiResponse searchLearnerAchievements(SearchCriteria searchCriteria, String authToken);
+}
diff --git a/src/main/java/com/igot/cb/profile/service/AchievementServiceImpl.java b/src/main/java/com/igot/cb/profile/service/AchievementServiceImpl.java
new file mode 100644
index 0000000..22cea5d
--- /dev/null
+++ b/src/main/java/com/igot/cb/profile/service/AchievementServiceImpl.java
@@ -0,0 +1,679 @@
+package com.igot.cb.profile.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.auth0.jwt.JWT;
+import com.auth0.jwt.algorithms.Algorithm;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.igot.cb.authentication.util.AccessTokenValidator;
+import com.igot.cb.transactional.cassandrautils.CassandraOperation;
+import com.igot.cb.transactional.elasticsearch.dto.SearchCriteria;
+import com.igot.cb.transactional.elasticsearch.dto.SearchResult;
+import com.igot.cb.transactional.elasticsearch.service.EsClientService;
+import com.igot.cb.transactional.redis.cache.CacheService;
+import com.igot.cb.util.ApiResponse;
+import com.igot.cb.util.CbServerProperties;
+import com.igot.cb.util.Constants;
+import com.igot.cb.util.ProjectUtil;
+import jakarta.annotation.PostConstruct;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.collections4.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDate;
+import java.util.*;
+
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+@Service
+@Slf4j
+public class AchievementServiceImpl implements AchievementService{
+
+    private List<String> requiredFields;
+
+    private final AccessTokenValidator accessTokenValidator;
+
+    private final CbServerProperties cbServerProperties;
+
+    private final CassandraOperation cassandraOperation;
+
+    private final EsClientService esClientService;
+
+    private final ObjectMapper objectMapper;
+
+    private final RedisTemplate<String, SearchResult> redisTemplate;
+
+    private static final String FIELD_REASON = "reason";
+
+    private static final String FIELD_LEARNER_ID = "learnerId";
+
+    @Autowired
+    private CacheService cacheService;
+
+    @Autowired
+    public AchievementServiceImpl(
+            AccessTokenValidator accessTokenValidator,
+            CbServerProperties cbServerProperties,
+            CassandraOperation cassandraOperation,
+            EsClientService esClientService,
+            ObjectMapper objectMapper,
+            @Qualifier(Constants.SEARCH_RESULT_REDIS_TEMPLATE) RedisTemplate<String, SearchResult> redisTemplate,
+            CacheService cacheService) {
+        this.accessTokenValidator = accessTokenValidator;
+        this.cbServerProperties = cbServerProperties;
+        this.cassandraOperation = cassandraOperation;
+        this.esClientService = esClientService;
+        this.objectMapper = objectMapper;
+        this.redisTemplate = redisTemplate;
+        this.cacheService = cacheService;
+    }
+
+    @PostConstruct
+    private void initRequiredFields() {
+        requiredFields = Arrays.asList(cbServerProperties.getRequiredFieldsProperty().split(","));
+    }
+
+    @Override
+    public ApiResponse createLearnerAchievement(Map<String, Object> request, String userToken, String orgId) {
+        ApiResponse response = ProjectUtil.createDefaultResponse("api.learnerAchievement.create");
+
+        Map<String, Object> requestData = (Map<String, Object>) request.get(Constants.REQUEST);
+        String userId = accessTokenValidator.fetchUserIdFromAccessToken(userToken);
+        if (StringUtils.isEmpty(userId)) {
+            ProjectUtil.errorResponse(response, "UserId not Found", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        String validationError = validateRequetData(requestData);
+        if (StringUtils.isNotBlank(validationError)) {
+            ProjectUtil.errorResponse(response, validationError, HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        String contextType = (String) requestData.get(Constants.CONTEXT_TYPE);
+        String source = (String) requestData.get(Constants.SOURCE);
+        Map<String, Object> contextData =
+                (Map<String, Object>) requestData.get(Constants.CONTEXT_DATA);
+
+        String id = UUID.randomUUID().toString();
+        LocalDate createdOn = LocalDate.now();
+
+        Map<String, Object> achievementRecord = new HashMap<>();
+        achievementRecord.put(Constants.USER_ID_RQST, userId);
+        achievementRecord.put(Constants.CONTEXT_TYPE, contextType);
+        achievementRecord.put(Constants.ID, id);
+        achievementRecord.put(Constants.ORG_ID, orgId);
+        achievementRecord.put(Constants.SOURCE, source);
+        achievementRecord.put(Constants.CONTEXT_DATA, contextData);
+        achievementRecord.put(Constants.STATUS, Constants.PENDING);
+        achievementRecord.put(Constants.CREATED_ON, createdOn);
+
+        // Save into Cassandra
+        boolean isSaved = saveAchievementToCassandra(achievementRecord);
+        if (!isSaved) {
+            ProjectUtil.errorResponse(response,
+                    "Failed to save learner achievement info",
+                    HttpStatus.INTERNAL_SERVER_ERROR);
+            return response;
+        }
+        Map<String, Object> esRecord = new HashMap<>(achievementRecord);
+        esRecord.put(Constants.CREATED_ON, createdOn.toString());
+        Map<String, Object> map = objectMapper.convertValue(esRecord, Map.class);
+        esClientService.addDocument(Constants.LEARNER_ACHIEVEMENT_INDEX, Constants.INDEX_TYPE, id, map, cbServerProperties.getAchievementEsRequiredFieldsMappingPath());
+
+        // Cache record
+        cacheService.putCache(
+                buildCacheKey("user:achievement", userId, contextType, id),
+                esRecord
+        );
+        response.setResponseCode(HttpStatus.OK);
+        response.setResponse(achievementRecord);
+        return response;
+    }
+
+    @Override
+    public ApiResponse updateLearnerAchievement(Map<String, Object> request, String userToken, String orgId) {
+        ApiResponse response = ProjectUtil.createDefaultResponse("api.learnerAchievement.update");
+        Map<String, Object> requestData = (Map<String, Object>) request.get(Constants.REQUEST);
+        String validateMessage = validateRequetData(requestData);
+        if (StringUtils.isNotBlank(validateMessage)) {
+            ProjectUtil.errorResponse(response, validateMessage, HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        String userId = accessTokenValidator.fetchUserIdFromAccessToken(userToken);
+        if (StringUtils.isEmpty(userId)) {
+            ProjectUtil.errorResponse(response, "UserId not Found", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        // Validate mandatory fields
+        String id = (String) requestData.get(Constants.ID);
+        String contextType = (String) requestData.get(Constants.CONTEXT_TYPE);
+
+        if (StringUtils.isBlank(id) || StringUtils.isBlank(contextType)) {
+            ProjectUtil.errorResponse(response, "id and contextType are mandatory", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        Map<String, Object> newContextData =
+                (Map<String, Object>) requestData.get(Constants.CONTEXT_DATA);
+
+        if (MapUtils.isEmpty(newContextData)) {
+            ProjectUtil.errorResponse(response, "contextData is mandatory for update", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+
+        //  Fetch existing record
+        Map<String, Object> existingRecord =
+                getAchievementFromCassandra(userId, contextType, id);
+
+        if (existingRecord == null) {
+            ProjectUtil.errorResponse(response, "Achievement records not found", HttpStatus.NOT_FOUND);
+            return response;
+        }
+
+        String currentStatus = (String) existingRecord.get(Constants.STATUS);
+
+        if (!Constants.PENDING.equalsIgnoreCase(currentStatus)) {
+            ProjectUtil.errorResponse(response,
+                    "Only PENDING achievements can be updated",
+                    HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        existingRecord.put(Constants.CONTEXT_DATA, newContextData);
+        LocalDate updateOn = LocalDate.now();
+        existingRecord.put(Constants.UPDATED_ON, updateOn);
+        existingRecord.put(Constants.UPDATED_BY, userId);
+
+        boolean isSaved = saveAchievementToCassandra(existingRecord);
+        if (!isSaved) {
+            ProjectUtil.errorResponse(response,
+                    "Failed to update learner achievement",
+                    HttpStatus.INTERNAL_SERVER_ERROR);
+            return response;
+        }
+        Map<String, Object> esRecord = new HashMap<>(existingRecord);
+        esRecord.put(Constants.CREATED_ON, existingRecord.get(Constants.CREATED_ON).toString());
+        esRecord.put(Constants.UPDATED_ON, updateOn.toString());
+        Map<String, Object> map = objectMapper.convertValue(esRecord, Map.class);
+        esClientService.updateDocument(Constants.LEARNER_ACHIEVEMENT_INDEX, Constants.INDEX_TYPE, id, map, cbServerProperties.getAchievementEsRequiredFieldsMappingPath());
+        cacheService.putCache(
+                buildCacheKey("user:achievement", userId, contextType, id),
+                esRecord
+        );
+        response.setResponseCode(HttpStatus.OK);
+        response.setResponse(esRecord);
+        return response;
+    }
+
+    @Override
+    public ApiResponse readLearnerAchievement(String achievementId, String userToken, String contextType) {
+        ApiResponse response = ProjectUtil.createDefaultResponse("api.learnerAchievement.read");
+        String userId = accessTokenValidator.fetchUserIdFromAccessToken(userToken);
+        if (StringUtils.isBlank(userId)) {
+            ProjectUtil.errorResponse(response, "UserId not Found", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        if (StringUtils.isBlank(achievementId)) {
+            ProjectUtil.errorResponse(response, "achievementId is mandatory", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        Map<String, Object> achievement = null;
+        if (StringUtils.isNotBlank(contextType)) {
+            String cacheKey = buildCacheKey("user:achievement", userId, contextType, achievementId);
+            achievement = getAchievementFromCache(cacheKey);
+        }
+        if (MapUtils.isEmpty(achievement)) {
+            achievement = getAndCacheAchievementFromCassandra(userId, contextType, achievementId);
+        }
+        if (MapUtils.isEmpty(achievement)) {
+            ProjectUtil.errorResponse(response, "Achievement not found", HttpStatus.NOT_FOUND);
+            return response;
+        }
+
+        response.setResponseCode(HttpStatus.OK);
+        response.setResponse(achievement);
+        return response;
+    }
+
+    @Override
+    public ApiResponse deleteLearnerAchievement(Map<String, Object> request, String userToken) {
+        ApiResponse response = ProjectUtil.createDefaultResponse("api.learnerAchievement.delete");
+        if (request == null || !(request.get(Constants.REQUEST) instanceof Map)) {
+            ProjectUtil.errorResponse(response, "Missing or invalid 'request' object in payload", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        Map<String, Object> reqMap = (Map<String, Object>) request.get(Constants.REQUEST);
+        String achievementId = (String) reqMap.get(Constants.ID);
+        String contextType = (String) reqMap.get(Constants.CONTEXT_TYPE);
+        if (StringUtils.isBlank(achievementId) || StringUtils.isBlank(contextType)) {
+            ProjectUtil.errorResponse(response, "achievementId and contextType are mandatory", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        String userId = accessTokenValidator.fetchUserIdFromAccessToken(userToken);
+        if (StringUtils.isBlank(userId)) {
+            ProjectUtil.errorResponse(response, "UserId not Found", HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        // Delete from Cassandra
+        Map<String, Object> compositeKey = new HashMap<>();
+        compositeKey.put(Constants.ID, achievementId);
+        compositeKey.put(Constants.USER_ID_RQST, userId);
+        compositeKey.put(Constants.CONTEXT_TYPE, contextType);
+        Map<String, Object> cassandraResponse = cassandraOperation.deleteRecordByCompositeKey(
+                Constants.KEYSPACE_SUNBIRD,
+                Constants.LEARNER_ACHIEVEMENT_TABLE,
+                compositeKey
+        );
+        if (!Constants.SUCCESS.equals(cassandraResponse.get(Constants.RESPONSE))) {
+            ProjectUtil.errorResponse(response, "Failed to delete achievement record", HttpStatus.INTERNAL_SERVER_ERROR);
+            return response;
+        }
+        // Remove from cache
+        String cacheKey = buildCacheKey("user:achievement", userId, contextType, achievementId);
+        cacheService.removeCache(cacheKey);
+        // Remove from ES
+        try {
+            esClientService.deleteDocument(achievementId, Constants.LEARNER_ACHIEVEMENT_INDEX);
+        } catch (Exception e) {
+            log.warn("Failed to delete achievement from ES for id {}", achievementId, e);
+        }
+        response.setResponseCode(HttpStatus.OK);
+        response.getResult().put("message", "Achievement deleted successfully");
+        return response;
+    }
+
+    @Override
+    public ApiResponse statusUpdateLearnerAchievement(Map<String, Object> request, String authToken) {
+        log.info("AchievementService::statusUpdateLearnerAchievement");
+        ApiResponse response = ProjectUtil.createDefaultResponse(Constants.API_ACHIEVEMENT_STATUS_UPDATE);
+        try {
+            String userIdFromToken = accessTokenValidator.fetchUserIdFromAccessToken(authToken);
+            if (StringUtils.isBlank(userIdFromToken)) {
+                ProjectUtil.errorResponse(response, "Invalid or missing access token", HttpStatus.UNAUTHORIZED);
+                return response;
+            }
+            if (!validateStatusUpdateRequest(request, response)) {
+                return response;
+            }
+            Map<String, Object> reqMap = (Map<String, Object>) request.get(Constants.REQUEST);
+            Map<String, Object> compositeKey = new HashMap<>();
+            compositeKey.put(Constants.ID, reqMap.get(Constants.ID));
+            compositeKey.put(Constants.USER_ID_LOWER, reqMap.get(FIELD_LEARNER_ID));
+            compositeKey.put(Constants.FIELD_CONTEXT_TYPE, reqMap.get(Constants.CONTEXT_TYPE_KEY));
+            List<Map<String, Object>> records = cassandraOperation.getAllRecordsByPrimaryKey(
+                Constants.KEYSPACE_SUNBIRD,
+                Constants.LEARNER_ACHIEVEMENT_TABLE,
+                compositeKey,
+                null,
+                Constants.CASSANDRA_FETCH_LIMIT
+            );
+            if (CollectionUtils.isEmpty(records)) {
+                ProjectUtil.errorResponse(response, "Achievement record not found for update", HttpStatus.NOT_FOUND);
+                return response;
+            }
+            // Additional validation: status must be PENDING
+            String currentStatus = String.valueOf(records.get(0).get(Constants.STATUS));
+            if (!Constants.PENDING.equalsIgnoreCase(currentStatus)) {
+                ProjectUtil.errorResponse(response, "Achievement status must be 'PENDING' to update. Current status: " + currentStatus, HttpStatus.BAD_REQUEST);
+                return response;
+            }
+            Map<String, Object> updateAttributes = new HashMap<>();
+            updateAttributes.put(Constants.STATUS, reqMap.get(Constants.STATUS));
+            updateAttributes.put(FIELD_REASON, reqMap.get(FIELD_REASON));
+            updateAttributes.put(Constants.FIELD_APPROVED_BY, userIdFromToken);
+            // Store approvedon as date (yyyy-MM-dd) for Cassandra
+            String approvedOnDate = java.time.LocalDate.now().toString();
+            updateAttributes.put(Constants.FIELD_APPROVED_ON, approvedOnDate);
+            Map<String, Object> cassandraResponse = cassandraOperation.updateRecordByCompositeKey(
+                Constants.KEYSPACE_SUNBIRD,
+                Constants.LEARNER_ACHIEVEMENT_TABLE,
+                updateAttributes,
+                compositeKey
+            );
+            if (!Constants.SUCCESS.equals(cassandraResponse.get(Constants.RESPONSE))) {
+                ProjectUtil.errorResponse(response, String.valueOf(cassandraResponse.get(Constants.ERROR_MESSAGE)), HttpStatus.INTERNAL_SERVER_ERROR);
+                return response;
+            }
+            updateAchievementInES(records, reqMap, userIdFromToken, approvedOnDate);
+            response.getResult().put("message", "Achievement status updated successfully");
+        } catch (Exception e) {
+            log.error("Exception in statusUpdateLearnerAchievement", e);
+            ProjectUtil.errorResponse(response, "Exception occurred: " + e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR);
+        }
+        return response;
+    }
+
+    private void updateAchievementInES(List<Map<String, Object>> records, Map<String, Object> reqMap, String userIdFromToken, String approvedOnDate) {
+        try {
+            Map<String, Object> esUpdateMap = new HashMap<>();
+            if (CollectionUtils.isNotEmpty(records)) {
+                Map<String, Object> dbRecord = records.get(0);
+                for (Map.Entry<String, Object> entry : dbRecord.entrySet()) {
+                    Object value = entry.getValue();
+                    if (value instanceof java.time.LocalDate) {
+                        esUpdateMap.put(entry.getKey(), value.toString());
+                    } else if (value instanceof java.time.LocalDateTime) {
+                        esUpdateMap.put(entry.getKey(), value.toString());
+                    } else if ("contextdata".equalsIgnoreCase(entry.getKey()) && value != null) {
+                        if (value instanceof String) {
+                            try {
+                                Map<String, Object> contextDataMap = objectMapper.readValue((String) value, Map.class);
+                                esUpdateMap.put(entry.getKey(), contextDataMap);
+                            } catch (Exception ex) {
+                                log.warn("Failed to parse contextData string to Map for ES. Storing as empty object.", ex);
+                                esUpdateMap.put(entry.getKey(), new HashMap<>());
+                            }
+                        } else if (value instanceof Map) {
+                            esUpdateMap.put(entry.getKey(), value);
+                        } else {
+                            esUpdateMap.put(entry.getKey(), new HashMap<>());
+                        }
+                    } else {
+                        esUpdateMap.put(entry.getKey(), value);
+                    }
+                }
+            }
+            esUpdateMap.put(Constants.STATUS, reqMap.get(Constants.STATUS));
+            esUpdateMap.put(FIELD_REASON, reqMap.get(FIELD_REASON));
+            esUpdateMap.put(Constants.FIELD_APPROVED_BY, userIdFromToken);
+            esUpdateMap.put(Constants.FIELD_APPROVED_ON, approvedOnDate);
+            esClientService.updateDocument(
+                Constants.LEARNER_ACHIEVEMENT_INDEX,
+                null,
+                String.valueOf(reqMap.get(Constants.ID)),
+                esUpdateMap,
+                cbServerProperties.getAchievementEsRequiredFieldsMappingPath()
+            );
+        } catch (Exception e) {
+            log.error("Exception while updating achievement in ES", e);
+        }
+    }
+
+    private boolean validateStatusUpdateRequest(Map<String, Object> request, ApiResponse response) {
+        if (MapUtils.isEmpty(request) || !(request.get(Constants.REQUEST) instanceof Map) || MapUtils.isEmpty((Map<?, ?>) request.get(Constants.REQUEST))) {
+            ProjectUtil.errorResponse(response, "Missing or invalid 'request' object in payload", HttpStatus.BAD_REQUEST);
+            return false;
+        }
+        Map<String, Object> reqMap = (Map<String, Object>) request.get(Constants.REQUEST);
+        for (String field : requiredFields) {
+            if (!reqMap.containsKey(field) || reqMap.get(field) == null) {
+                ProjectUtil.errorResponse(response, "Missing required field: " + field, HttpStatus.BAD_REQUEST);
+                return false;
+            }
+        }
+        String statusValue = String.valueOf(reqMap.get(Constants.STATUS));
+        if (!Constants.APPROVED_KEY.equalsIgnoreCase(statusValue) && !Constants.REJECTED.equalsIgnoreCase(statusValue)) {
+            ProjectUtil.errorResponse(response, "Invalid status value. Allowed values are 'Approved' or 'Reject'", HttpStatus.BAD_REQUEST);
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public ApiResponse searchLearnerAchievements(SearchCriteria searchCriteria, String authToken) {
+        log.info("AchievementService::searchLearnerAchievements");
+        ApiResponse response = ProjectUtil.createDefaultResponse(Constants.API_ACHIEVEMENT_SEARCH);
+        String cacheKey = generateRedisJwtTokenKey(searchCriteria);
+        SearchResult searchResult = redisTemplate.opsForValue().get(cacheKey);
+        if (searchResult != null) {
+            log.info("DiscussionServiceImpl::searchDiscussion:  search result fetched from redis");
+            response.getResult().put(Constants.SEARCH_RESULTS, searchResult);
+            return response;
+        }
+        String searchString = searchCriteria.getSearchString();
+        if (searchString != null && !searchString.isEmpty() && searchString.length() < 3) {
+            ProjectUtil.errorResponse(response, Constants.MINIMUM_CHARACTERS_NEEDED, HttpStatus.BAD_REQUEST);
+            return response;
+        }
+        try {
+            log.info("DiscussionServiceImpl::searchDiscussion:  search result fetched from es");
+            if (MapUtils.isEmpty(searchCriteria.getFilterCriteriaMap())) {
+                searchCriteria.setFilterCriteriaMap(new HashMap<>());
+            }
+            searchResult = esClientService.searchDocuments(Constants.LEARNER_ACHIEVEMENT_INDEX, searchCriteria);
+            if (CollectionUtils.isEmpty(searchResult.getData())) {
+                ProjectUtil.errorResponse(response, Constants.NO_DATA_FOUND, HttpStatus.OK);
+                response.getResult().put(Constants.SEARCH_RESULTS, searchResult);
+                return response;
+            }
+            List<Map<String, Object>> achievemnets = searchResult.getData();
+            searchResult.setUserDetails(fetchUsernamesFromSearchData(achievemnets));
+            searchResult.setData(achievemnets);
+            redisTemplate.opsForValue().set(cacheKey, searchResult, cbServerProperties.getSearchResultRedisTtl(), TimeUnit.SECONDS);
+            response.getResult().put(Constants.SEARCH_RESULTS, searchResult);
+            return response;
+        } catch (Exception e) {
+            log.error("error while searching discussion : {} .", e.getMessage(), e);
+            ProjectUtil.errorResponse(response, e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR);
+            return response;
+        }
+    }
+
+    /**
+     * Extracts unique userIds from search result data and fetches usernames for each.
+     * @param data List of search result maps (each representing a record)
+     * @return Map of userId to username
+     */
+    private Map<String, String> fetchUsernamesFromSearchData(List<Map<String, Object>> data) {
+        Set<String> uniqueUserIds = new HashSet<>();
+        for (Map<String, Object> item : data) {
+            Object userIdObj = item.get(Constants.USER_ID);
+            if (StringUtils.isEmpty((String) userIdObj)) userIdObj = item.get(Constants.USER_ID_LOWER);
+            if (userIdObj instanceof String && StringUtils.isNotBlank((String) userIdObj)) {
+                uniqueUserIds.add((String) userIdObj);
+            }
+        }
+        // Fetch user details (replace with actual Redis/Cassandra logic)
+        List<Object> userDetailsList = fetchUserDetails(new ArrayList<>(uniqueUserIds));
+        Map<String, String> userIdToUsername = new HashMap<>();
+        for (Object user : userDetailsList) {
+            if (user instanceof Map) {
+                Map userMap = (Map) user;
+                Object idObj = userMap.get(Constants.USER_ID_KEY);
+                Object nameObj = userMap.get(Constants.FIRST_NAME_KEY);
+                if (idObj instanceof String && nameObj instanceof String) {
+                    userIdToUsername.put((String) idObj, (String) nameObj);
+                }
+            }
+        }
+        return userIdToUsername;
+    }
+
+    private List<Object> fetchUserDetails(List<String> userIds) {
+        // Prepare Redis keys (assuming prefix is needed)
+        List<String> redisKeys = userIds.stream()
+            .map(id -> Constants.USER_PREFIX + id)
+            .collect(Collectors.toList());
+        // Fetch values for all keys from Redis
+        List<Object> redisResults = cacheService.hget(redisKeys); // Use your cacheService
+        // Build userDetailsMap from redis results
+        Map<String, Object> userDetailsMap = redisResults.stream()
+                .filter(Objects::nonNull)
+                .map(user -> (Map<String, Object>) user)
+                .filter(user -> user.get(Constants.USER_ID_KEY) != null)
+                .collect(Collectors.toMap(
+                        user -> user.get(Constants.USER_ID_KEY).toString(),
+                        user -> user,
+                        (u1, u2) -> u1));
+        // Find missing userIds
+        List<String> missingUserIds = userIds.stream()
+                .filter(id -> !userDetailsMap.containsKey(id))
+                .collect(Collectors.toList());
+        // Fetch from Cassandra if missing
+        if (!missingUserIds.isEmpty()) {
+            List<Object> cassandraResults = fetchUserFromPrimary(missingUserIds);
+            userDetailsMap.putAll(cassandraResults.stream()
+                    .map(user -> (Map<String, Object>) user)
+                    .filter(user -> user.get(Constants.USER_ID_KEY) != null)
+                    .collect(Collectors.toMap(
+                            user -> user.get(Constants.USER_ID_KEY).toString(),
+                            user -> user,
+                            (u1, u2) -> u1)));
+        }
+        return new ArrayList<>(userDetailsMap.values());
+    }
+
+    public List<Object> fetchUserFromPrimary(List<String> userIds) {
+        log.info("AchievementServiceImpl::fetchUserFromPrimary: Fetching user data from Cassandra");
+        List<Object> userList = new ArrayList<>();
+        Map<String, Object> propertyMap = new HashMap<>();
+        propertyMap.put(Constants.ID, userIds);
+        long startTime = System.currentTimeMillis();
+        List<Map<String, Object>> userInfoList = cassandraOperation.getRecordsByPropertiesWithoutFiltering(
+                Constants.KEYSPACE_SUNBIRD, Constants.USER_TABLE, propertyMap,
+                Arrays.asList(Constants.FIRST_NAME, Constants.ID), null);
+        // updateMetricsDbOperation(Constants.DISCUSSION_SEARCH, Constants.CASSANDRA, Constants.READ, startTime);
+        userList = userInfoList.stream()
+                .map(userInfo -> {
+                    Map<String, Object> userMap = new HashMap<>();
+                    String userId = (String) userInfo.get(Constants.ID);
+                    String userName = (String) userInfo.get(Constants.FIRST_NAME_CAMEL_CASE);
+                    userMap.put(Constants.USER_ID_KEY, userId);
+                    userMap.put(Constants.FIRST_NAME_KEY, userName);
+                    return userMap;
+                })
+                .collect(Collectors.toList());
+        return userList;
+    }
+
+    public String generateRedisJwtTokenKey(Object requestPayload) {
+        if (requestPayload != null) {
+            try {
+                String reqJsonString = objectMapper.writeValueAsString(requestPayload);
+                return JWT.create()
+                        .withClaim(Constants.REQUEST, reqJsonString)
+                        .sign(Algorithm.HMAC256(Constants.JWT_SECRET_KEY));
+            } catch (JsonProcessingException e) {
+                log.error("Error occurred while converting json object to json string", e);
+            }
+        }
+        return "";
+    }
+
+    private String validateRequetData(Map<String, Object> requestData) {
+        String requestContextType = (String) requestData.get(Constants.CONTEXT_TYPE);
+        String[] configuredContextType = cbServerProperties.getContextType();
+        if (StringUtils.isBlank(requestContextType)) {
+            return "contextType is missing in request";
+        }
+
+        boolean isValid = Arrays.stream(configuredContextType)
+                .anyMatch(ct -> ct.equalsIgnoreCase(requestContextType));
+        if (!isValid) {
+            return "Invalid contextType. Allowed values: " + String.join(",", configuredContextType);
+        }
+        Map<String, Object> contextData =
+                (Map<String, Object>) requestData.get(Constants.CONTEXT_DATA);
+
+        if (MapUtils.isEmpty(contextData)|| contextData.isEmpty()) {
+            return "contextData is missing in request";
+        }
+        String requiredFieldsConfig = cbServerProperties.getAchievementsMandatoryFields();
+        String[] requiredFields = requiredFieldsConfig.split(",");
+
+        //  Validate mandatory fields
+        for (String field : requiredFields) {
+            Object value = contextData.get(field.trim());
+            if (value == null) {
+                return field + " is mandatory and missing";
+            }
+            if (value instanceof String && StringUtils.isBlank((String) value)) {
+                return field + " is mandatory and cannot be empty";
+            }
+        }
+        return null;
+    }
+
+    private boolean saveAchievementToCassandra(Map<String, Object> achievementRecord) {
+        try {
+            Map<String, Object> query = new HashMap<>(achievementRecord);
+            String contextDataJson = objectMapper.writeValueAsString(
+                    achievementRecord.get(Constants.CONTEXT_DATA)
+            );
+            query.put(Constants.CONTEXT_DATA, contextDataJson);
+
+            ApiResponse insertResponse = (ApiResponse) cassandraOperation.insertRecord(
+                    Constants.KEYSPACE_SUNBIRD,
+                    Constants.LEARNER_ACHIEVEMENT_TABLE,
+                    query
+            );
+            return Constants.SUCCESS.equalsIgnoreCase(
+                    (String) insertResponse.get(Constants.RESPONSE)
+            );
+        } catch (Exception e) {
+            log.error("Failed to insert learner achievement", e);
+            return false;
+        }
+    }
+
+    private String buildCacheKey(String prefix, String userId, String contextType, String id) {
+        return String.join(":", prefix, contextType, userId, id);
+    }
+
+    private Map<String, Object> getAchievementFromCassandra(String userId, String contextType, String id) {
+        try {
+            Map<String, Object> propertyMap = new HashMap<>();
+            propertyMap.put(Constants.USER_ID_RQST, userId);
+            propertyMap.put(Constants.CONTEXT_TYPE, contextType);
+            propertyMap.put(Constants.ID, id);
+            List<String> fields = new ArrayList<>();
+
+            List<Map<String, Object>> result =
+                    cassandraOperation.getRecordsByPropertiesByKey(
+                            Constants.KEYSPACE_SUNBIRD,
+                            Constants.LEARNER_ACHIEVEMENT_TABLE,
+                            propertyMap,
+                            fields,
+                            Constants.USERID_KEY
+                    );
+            if (result == null || result.isEmpty()) {
+                return null;
+            }
+            Map<String, Object> record = result.get(0);
+
+            // Convert contextdata JSON string back to Map
+            Object contextDataObj = record.get(Constants.CONTEXT_DATA);
+            if (contextDataObj instanceof String) {
+                Map<String, Object> contextData =
+                        objectMapper.readValue((String) contextDataObj, Map.class);
+                record.put(Constants.CONTEXT_DATA, contextData);
+            }
+            return record;
+
+        } catch (Exception e) {
+            log.error("Failed to fetch learner achievement for userId={}, contextType={}, id={}",
+                    userId, contextType, id, e);
+            return null;
+        }
+    }
+
+    private Map<String, Object> getAchievementFromCache(String cacheKey) {
+        String cachedJson = cacheService.getCache(cacheKey);
+        if (StringUtils.isNotBlank(cachedJson)) {
+            try {
+                log.info("reading from cache {}", cacheKey);
+                return objectMapper.readValue(cachedJson, new com.fasterxml.jackson.core.type.TypeReference<Map<String, Object>>() {
+                });
+            } catch (Exception e) {
+                log.error("Failed to deserialize cached achievement for key {}", cacheKey, e);
+            }
+        }
+        return null;
+    }
+
+    private Map<String, Object> getAndCacheAchievementFromCassandra(String userId, String contextType, String achievementId) {
+        Map<String, Object> achievement = getAchievementFromCassandra(userId, contextType, achievementId);
+        if (achievement != null && StringUtils.isNotBlank(contextType)) {
+            try {
+                String achievementJson = objectMapper.writeValueAsString(achievement);
+                cacheService.putCache(buildCacheKey("user:achievement", userId, contextType, achievementId), achievementJson);
+            } catch (Exception e) {
+                log.error("Failed to serialize achievement for caching", e);
+            }
+        }
+        return achievement;
+    }
+}
diff --git a/src/main/java/com/igot/cb/transactional/cassandrautils/CassandraOperation.java b/src/main/java/com/igot/cb/transactional/cassandrautils/CassandraOperation.java
index 0efb814..4709c76 100644
--- a/src/main/java/com/igot/cb/transactional/cassandrautils/CassandraOperation.java
+++ b/src/main/java/com/igot/cb/transactional/cassandrautils/CassandraOperation.java
@@ -49,4 +49,6 @@ public Map<String, Object> updateRecordByCompositeKey(
 
         public List<Map<String, Object>> getAllRecordsByPrimaryKey(String keyspaceName, String tableName,
                         Map<String, Object> primaryKey, List<String> fields, int pageSize);
+
+        public Map<String, Object> deleteRecordByCompositeKey(String keyspaceName, String tableName, Map<String, Object> compositeKey);
 }
diff --git a/src/main/java/com/igot/cb/transactional/cassandrautils/CassandraOperationImpl.java b/src/main/java/com/igot/cb/transactional/cassandrautils/CassandraOperationImpl.java
index c990ee7..db60961 100644
--- a/src/main/java/com/igot/cb/transactional/cassandrautils/CassandraOperationImpl.java
+++ b/src/main/java/com/igot/cb/transactional/cassandrautils/CassandraOperationImpl.java
@@ -264,4 +264,34 @@ public List<Map<String, Object>> getAllRecordsByPrimaryKey(String keyspaceName,
         }
         return allResults;
     }
+
+    @Override
+    public Map<String, Object> deleteRecordByCompositeKey(String keyspaceName, String tableName, Map<String, Object> compositeKey) {
+        Map<String, Object> response = new HashMap<>();
+        try {
+            StringBuilder queryBuilder = new StringBuilder();
+            queryBuilder.append("DELETE FROM ")
+                        .append(keyspaceName).append(".").append(tableName)
+                        .append(" WHERE ");
+            List<Object> values = new ArrayList<>();
+            int count = 0;
+            for (Map.Entry<String, Object> entry : compositeKey.entrySet()) {
+                if (count > 0) queryBuilder.append(" AND ");
+                queryBuilder.append(entry.getKey()).append(" = ?");
+                values.add(entry.getValue());
+                count++;
+            }
+            String query = queryBuilder.toString();
+            CqlSession session = connectionManager.getSession(keyspaceName);
+            PreparedStatement statement = session.prepare(query);
+            BoundStatement boundStatement = statement.bind(values.toArray());
+            session.execute(boundStatement);
+            response.put(Constants.RESPONSE, Constants.SUCCESS);
+        } catch (Exception e) {
+            logger.error("Error deleting record from {}: {}", tableName, e.getMessage());
+            response.put(Constants.RESPONSE, Constants.FAILED);
+            response.put(Constants.ERROR_MESSAGE, e.getMessage());
+        }
+        return response;
+    }
 }
diff --git a/src/main/java/com/igot/cb/transactional/elasticsearch/config/EsClientConfig.java b/src/main/java/com/igot/cb/transactional/elasticsearch/config/EsClientConfig.java
new file mode 100644
index 0000000..fa1cdee
--- /dev/null
+++ b/src/main/java/com/igot/cb/transactional/elasticsearch/config/EsClientConfig.java
@@ -0,0 +1,53 @@
+package com.igot.cb.transactional.elasticsearch.config;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import co.elastic.clients.transport.ElasticsearchTransport;
+import co.elastic.clients.transport.rest_client.RestClientTransport;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpResponseInterceptor;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class EsClientConfig {
+
+    @Value("${elasticsearch.host}")
+    private String elasticsearchHost;
+
+    @Value("${elasticsearch.port}")
+    private int elasticsearchPort;
+
+    @Value("${elasticsearch.username}")
+    private String elasticsearchUsername;
+
+    @Value("${elasticsearch.password}")
+    private String elasticsearchPassword;
+
+    @Bean
+    public ElasticsearchClient elasticsearchClient() {
+        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+        credentialsProvider.setCredentials(AuthScope.ANY,
+                new UsernamePasswordCredentials(elasticsearchUsername, elasticsearchPassword));
+
+        RestClientBuilder builder = RestClient.builder(
+                        new HttpHost(elasticsearchHost, elasticsearchPort, "http"))
+                .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider).addInterceptorLast((HttpResponseInterceptor) (response, context) ->
+                        response.addHeader("X-Elastic-Product", "Elasticsearch")))
+                .setDefaultHeaders(new org.apache.http.Header[]{
+                        new org.apache.http.message.BasicHeader("Content-Type", "application/json"),
+                        new org.apache.http.message.BasicHeader("X-Elastic-Product", "Elasticsearch")});
+        RestClient restClient = builder.build();
+        ElasticsearchTransport elasticsearchTransport = new RestClientTransport(restClient, new JacksonJsonpMapper());
+        ElasticsearchClient client = new ElasticsearchClient(elasticsearchTransport);
+        return client;
+    }
+
+}
diff --git a/src/main/java/com/igot/cb/transactional/elasticsearch/dto/FacetDTO.java b/src/main/java/com/igot/cb/transactional/elasticsearch/dto/FacetDTO.java
new file mode 100644
index 0000000..ae05ad9
--- /dev/null
+++ b/src/main/java/com/igot/cb/transactional/elasticsearch/dto/FacetDTO.java
@@ -0,0 +1,19 @@
+package com.igot.cb.transactional.elasticsearch.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.io.Serializable;
+
+@Getter
+@Setter
+@AllArgsConstructor
+@NoArgsConstructor
+public class FacetDTO implements Serializable {
+
+  private String value;
+
+  private Long count;
+}
diff --git a/src/main/java/com/igot/cb/transactional/elasticsearch/dto/SearchCriteria.java b/src/main/java/com/igot/cb/transactional/elasticsearch/dto/SearchCriteria.java
new file mode 100644
index 0000000..b1e08a4
--- /dev/null
+++ b/src/main/java/com/igot/cb/transactional/elasticsearch/dto/SearchCriteria.java
@@ -0,0 +1,39 @@
+package com.igot.cb.transactional.elasticsearch.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Getter
+@Setter
+@AllArgsConstructor
+@NoArgsConstructor
+public class SearchCriteria {
+
+    private HashMap<String, Object> filterCriteriaMap;
+
+    private List<String> requestedFields;
+
+    private int pageNumber;
+
+    private int pageSize;
+
+    private String orderBy;
+
+    private String orderDirection;
+
+    private String searchString;
+
+    private List<String> facets;
+
+    private Map<String, Object> query;
+
+    private String startsWith;
+
+    private String startsWithField;
+}
diff --git a/src/main/java/com/igot/cb/transactional/elasticsearch/dto/SearchResult.java b/src/main/java/com/igot/cb/transactional/elasticsearch/dto/SearchResult.java
new file mode 100644
index 0000000..366452a
--- /dev/null
+++ b/src/main/java/com/igot/cb/transactional/elasticsearch/dto/SearchResult.java
@@ -0,0 +1,22 @@
+package com.igot.cb.transactional.elasticsearch.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+@Getter
+@Setter
+@AllArgsConstructor
+@NoArgsConstructor
+public class SearchResult implements Serializable {
+
+  private List<Map<String, Object>> data;
+  private Map<String, List<FacetDTO>> facets;
+  private long totalCount;
+  private Map<String, String> userDetails;
+}
diff --git a/src/main/java/com/igot/cb/transactional/elasticsearch/service/EsClientService.java b/src/main/java/com/igot/cb/transactional/elasticsearch/service/EsClientService.java
new file mode 100644
index 0000000..6e45da9
--- /dev/null
+++ b/src/main/java/com/igot/cb/transactional/elasticsearch/service/EsClientService.java
@@ -0,0 +1,29 @@
+package com.igot.cb.transactional.elasticsearch.service;
+
+import co.elastic.clients.elasticsearch._types.query_dsl.Query;
+import co.elastic.clients.elasticsearch.core.BulkResponse;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.igot.cb.transactional.elasticsearch.dto.SearchCriteria;
+import com.igot.cb.transactional.elasticsearch.dto.SearchResult;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+
+public interface EsClientService {
+  String addDocument(String esIndexName, String type, String id, Map<String, Object> document, String JsonFilePath);
+
+  void updateDocument(String index, String indexType, String entityId, Map<String, Object> document, String JsonFilePath);
+
+  void deleteDocument(String documentId, String esIndexName);
+
+  void deleteDocumentsByCriteria(String esIndexName, Query query);
+
+  SearchResult searchDocuments(String esIndexName, SearchCriteria searchCriteria) throws Exception;
+
+  boolean isIndexPresent(String indexName);
+
+  BulkResponse saveAll(String esIndexName, List<JsonNode> entities) throws IOException;
+
+}
diff --git a/src/main/java/com/igot/cb/transactional/elasticsearch/service/EsClientServiceImpl.java b/src/main/java/com/igot/cb/transactional/elasticsearch/service/EsClientServiceImpl.java
new file mode 100644
index 0000000..293d71a
--- /dev/null
+++ b/src/main/java/com/igot/cb/transactional/elasticsearch/service/EsClientServiceImpl.java
@@ -0,0 +1,563 @@
+package com.igot.cb.transactional.elasticsearch.service;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.elasticsearch._types.FieldValue;
+import co.elastic.clients.elasticsearch._types.Refresh;
+import co.elastic.clients.elasticsearch._types.SortOptions;
+import co.elastic.clients.elasticsearch._types.SortOrder;
+import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
+import co.elastic.clients.elasticsearch._types.aggregations.Aggregation;
+import co.elastic.clients.elasticsearch._types.aggregations.StringTermsBucket;
+import co.elastic.clients.elasticsearch._types.aggregations.TermsAggregation;
+import co.elastic.clients.elasticsearch._types.query_dsl.*;
+import co.elastic.clients.elasticsearch.core.*;
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
+import co.elastic.clients.elasticsearch.core.search.Hit;
+import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
+import co.elastic.clients.elasticsearch.core.search.SourceConfig;
+import co.elastic.clients.elasticsearch.indices.GetIndexRequest;
+import co.elastic.clients.elasticsearch.indices.GetIndexResponse;
+import co.elastic.clients.elasticsearch.indices.RefreshRequest;
+import co.elastic.clients.json.JsonData;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.igot.cb.exceptions.CustomException;
+import com.igot.cb.transactional.elasticsearch.config.EsClientConfig;
+import com.igot.cb.transactional.elasticsearch.dto.FacetDTO;
+import com.igot.cb.transactional.elasticsearch.dto.SearchCriteria;
+import com.igot.cb.transactional.elasticsearch.dto.SearchResult;
+import com.igot.cb.util.CbServerProperties;
+import com.igot.cb.util.Constants;
+import com.networknt.schema.JsonSchemaFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.elasticsearch.client.RequestOptions;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Service;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.*;
+import java.util.Map.Entry;
+import java.util.stream.Collectors;
+
+@Service
+@Slf4j
+public class EsClientServiceImpl implements EsClientService {
+
+    /*@Autowired
+    private RestHighLevelClient elasticsearchClient;*/
+    private final EsClientConfig esConfig;
+    private final ElasticsearchClient elasticsearchClient;
+
+    @Autowired
+    private ObjectMapper objectMapper;
+    @Autowired
+    private CbServerProperties cbServerProperties;
+    @Autowired
+    private CbServerProperties serverConfig;
+
+    @Autowired
+    public EsClientServiceImpl(ElasticsearchClient elasticsearchClient, EsClientConfig esConnection) {
+        this.elasticsearchClient = elasticsearchClient;
+        this.esConfig = esConnection;
+    }
+
+
+    @Override
+    public String addDocument(
+            String esIndexName, String type, String id, Map<String, Object> document, String JsonFilePath) {
+        log.info("EsUtilServiceImpl :: addDocument");
+        try {
+            JsonSchemaFactory schemaFactory = JsonSchemaFactory.getInstance();
+            InputStream schemaStream = schemaFactory.getClass().getResourceAsStream(JsonFilePath);
+            Map<String, Object> map = objectMapper.readValue(schemaStream,
+                    new TypeReference<Map<String, Object>>() {
+                    });
+            Iterator<Entry<String, Object>> iterator = document.entrySet().iterator();
+            while (iterator.hasNext()) {
+                Entry<String, Object> entry = iterator.next();
+                String key = entry.getKey();
+                if (!map.containsKey(key)) {
+                    iterator.remove();
+                }
+            }
+            IndexRequest<Map<String,Object>> indexRequest = new IndexRequest.Builder<Map<String, Object>>()
+                    .index(esIndexName)
+                    .id(id)
+                    .document(document)
+                    .refresh(Refresh.True)
+                    .build();
+            IndexResponse response = elasticsearchClient.index(indexRequest);
+            return "Successfully indexed document with id: " + response.result();
+        } catch (Exception e) {
+            log.error("Issue while Indexing to es: {}", e.getMessage());
+            return null;
+        }
+    }
+
+    public void updateDocument(
+            String index, String indexType, String entityId, Map<String, Object> updatedDocument, String JsonFilePath) {
+        try {
+            JsonSchemaFactory schemaFactory = JsonSchemaFactory.getInstance();
+            InputStream schemaStream = schemaFactory.getClass().getResourceAsStream(JsonFilePath);
+            Map<String, Object> map = objectMapper.readValue(schemaStream,
+                    new TypeReference<Map<String, Object>>() {
+                    });
+            Iterator<Entry<String, Object>> iterator = updatedDocument.entrySet().iterator();
+            while (iterator.hasNext()) {
+                Entry<String, Object> entry = iterator.next();
+                String key = entry.getKey();
+                if (!map.containsKey(key)) {
+                    iterator.remove();
+                }
+            }
+            IndexRequest<Map<String, Object>> indexRequest = new IndexRequest.Builder<Map<String, Object>>()
+                    .index(index)
+                    .id(entityId)
+                    .document(updatedDocument)
+                    .refresh(Refresh.True)
+                    .build();
+            IndexResponse response = elasticsearchClient.index(indexRequest);
+        } catch (IOException e) {
+            log.error("Error occurred during deleting document in elasticsearch");
+        }
+    }
+
+    @Override
+    public void deleteDocument(String documentId, String esIndexName) {
+        try {
+            DeleteRequest request = new DeleteRequest.Builder().index(esIndexName).id(documentId).build();
+            DeleteResponse response = elasticsearchClient.delete(request);
+            if (response.result().jsonValue().equalsIgnoreCase("DELETED")) {
+                log.info("Document deleted successfully from elasticsearch.");
+                RefreshRequest refreshRequest = new RefreshRequest.Builder().index(esIndexName).build();
+                elasticsearchClient.indices().refresh(refreshRequest);
+                log.info("Index refreshed to reflect the document deletion.");
+            } else {
+                log.error("Document not found or failed to delete from elasticsearch.");
+            }
+        } catch (Exception e) {
+            log.error("Error occurred during deleting document in elasticsearch");
+        }
+    }
+
+    @Override
+    public SearchResult searchDocuments(String esIndexName, SearchCriteria searchCriteria) {
+        SearchRequest.Builder searchRequestBuilder = buildSearchRequest(searchCriteria);
+        assert searchRequestBuilder != null;
+        searchRequestBuilder.index(esIndexName);
+        try {
+            if (searchCriteria != null) {
+                int pageNumber = searchCriteria.getPageNumber();
+                int pageSize = searchCriteria.getPageSize();
+                int from = pageNumber * pageSize;
+                searchRequestBuilder.from(from);
+                if (pageSize > 0) {
+                    searchRequestBuilder.size(pageSize);
+                }
+
+            }
+
+            SearchRequest searchRequest = searchRequestBuilder.build();
+            log.info("Final search query: {}", searchRequest.toString());
+            SearchResponse<Object> paginatedSearchResponse =
+                    elasticsearchClient.search(searchRequest, Object.class);
+            List<Map<String, Object>> paginatedResult = extractPaginatedResult(paginatedSearchResponse);
+            Map<String, List<FacetDTO>> fieldAggregations =
+                    extractFacetData(paginatedSearchResponse, searchCriteria);
+            SearchResult searchResult = new SearchResult();
+            searchResult.setData(paginatedResult);
+            searchResult.setFacets(fieldAggregations);
+            searchResult.setTotalCount(paginatedSearchResponse.hits().total().value());
+            return searchResult;
+        } catch (IOException e) {
+            log.error("Error while fetching details from elastic search");
+            return null;
+        }
+    }
+
+    private Map<String, List<FacetDTO>> extractFacetData(
+            SearchResponse<Object> searchResponse, SearchCriteria searchCriteria) {
+        Map<String, List<FacetDTO>> fieldAggregations = new HashMap<>();
+        if (searchCriteria.getFacets() != null) {
+            for (String field : searchCriteria.getFacets()) {
+                Aggregate aggregate = searchResponse
+                        .aggregations()
+                        .get(field + "_agg");
+                if (aggregate.isSterms()) {
+                    List<FacetDTO> fieldValueList = new ArrayList<>();
+                    for (StringTermsBucket bucket : aggregate.sterms().buckets().array()) {
+                        if (!bucket.key().stringValue().isEmpty()) {
+                            FacetDTO facetDTO = new FacetDTO(bucket.key().stringValue(), bucket.docCount());
+                            fieldValueList.add(facetDTO);
+                        }
+                    }
+                    fieldAggregations.put(field, fieldValueList);
+                }
+            }
+        }
+        return fieldAggregations;
+    }
+
+    private List<Map<String, Object>> extractPaginatedResult(SearchResponse<Object> paginatedSearchResponse) {
+        List<Map<String, Object>> paginatedResult = new ArrayList<>();
+        for (Hit<Object> hit : paginatedSearchResponse.hits().hits()) {
+            paginatedResult.add((Map<String, Object>) hit.source());
+        }
+        return paginatedResult;
+    }
+
+    private SearchRequest.Builder buildSearchRequest(SearchCriteria searchCriteria) {
+        log.info("Building search query");
+        if (searchCriteria == null || searchCriteria.toString().isEmpty()) {
+            log.error("Search criteria body is missing");
+            return null;
+        }
+        BoolQuery.Builder boolQueryBuilder = buildFilterQuery(searchCriteria.getFilterCriteriaMap());
+        // Add startsWith logic if present
+        String startsWith = searchCriteria.getStartsWith();
+        String startsWithField = searchCriteria.getStartsWithField();
+        if (startsWith != null && !startsWith.trim().isEmpty() &&
+                startsWithField != null && !startsWithField.trim().isEmpty()) {
+            Query prefixQuery = Query.of(q -> q.prefix(p -> p
+                    .field(startsWithField + ".keyword")
+                    .value(startsWith)
+            ));
+            boolQueryBuilder.must(prefixQuery);
+        }
+        SearchRequest.Builder searchSourceBuilder = new SearchRequest.Builder();
+        searchSourceBuilder.query(boolQueryBuilder.build()._toQuery());
+        addSortToSearchSourceBuilder(searchCriteria, searchSourceBuilder);
+        addRequestedFieldsToSearchSourceBuilder(searchCriteria, searchSourceBuilder);
+        addQueryStringToFilter(searchCriteria.getSearchString(), boolQueryBuilder);
+        addFacetsToSearchSourceBuilder(searchCriteria.getFacets(), searchSourceBuilder);
+        Query queryPart = buildQueryPart(searchCriteria.getQuery());
+        boolQueryBuilder.must(queryPart);
+        return searchSourceBuilder;
+    }
+
+    private BoolQuery.Builder buildFilterQuery(Map<String, Object> filterCriteriaMap) {
+        BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
+        List<Query> mustNotQueries = new ArrayList<>();
+        List<Query> boolQueries = new ArrayList<>();
+
+        if (filterCriteriaMap != null) {
+            filterCriteriaMap.forEach(
+                    (field, value) -> {
+                        if (field.equals("must_not") && value instanceof ArrayList) {
+                            mustNotQueries.forEach(mustNotQuery -> boolQueryBuilder.mustNot(mustNotQuery));
+                        } else if (value instanceof Boolean) {
+                            boolQueries.add(Query.of(q ->q.term(t->t.field(field).value((boolean)value))));
+                        } else if (value instanceof ArrayList) {
+                            List<FieldValue> termsList = ((ArrayList<String>) value).stream()
+                                    .map(FieldValue::of)
+                                    .collect(Collectors.toList());
+                            boolQueryBuilder.must(Query.of(q -> q.terms(t -> t.field(field + Constants.KEYWORD).terms(terms -> terms.value(termsList)))));
+                        } else if (value instanceof String) {
+                            boolQueryBuilder.must(Query.of(q -> q.terms(t ->
+                                    t.field(field + Constants.KEYWORD)
+                                            .terms(terms -> terms.value(List.of(FieldValue.of((String) value))))
+                            )));
+                        } else if (value instanceof Map) {
+                            Map<String, Object> nestedMap = (Map<String, Object>) value;
+                            if (isRangeQuery(nestedMap)) {
+                                // Handle range query
+                                BoolQuery.Builder rangeOrNullQuery = QueryBuilders.bool();
+                                RangeQuery.Builder rangeQuery = QueryBuilders.range().field(field);
+                                nestedMap.forEach((rangeOperator, rangeValue) -> {
+                                    switch (rangeOperator) {
+                                        case Constants.SEARCH_OPERATION_GREATER_THAN_EQUALS:
+                                            rangeQuery.gte(JsonData.of(rangeValue));
+                                            break;
+                                        case Constants.SEARCH_OPERATION_LESS_THAN_EQUALS:
+                                            rangeQuery.lte(JsonData.of( rangeValue));
+                                            break;
+                                        case Constants.SEARCH_OPERATION_GREATER_THAN:
+                                            rangeQuery.gt(JsonData.of( rangeValue));
+                                            break;
+                                        case Constants.SEARCH_OPERATION_LESS_THAN:
+                                            rangeQuery.lt(JsonData.of( rangeValue));
+                                            break;
+                                    }
+                                });
+                                rangeOrNullQuery.should(rangeQuery.build()._toQuery());
+                                rangeOrNullQuery.should(Query.of(q -> q.bool(b -> b.mustNot(Query.of(qn -> qn.exists(e -> e.field(field)))))));
+                                boolQueryBuilder.must(rangeOrNullQuery.build()._toQuery());
+                            } else {
+                                nestedMap.forEach((nestedField, nestedValue) -> {
+                                    String fullPath = field + "." + nestedField;
+                                    if (nestedValue instanceof Boolean) {
+                                        boolQueryBuilder.must(Query.of(q -> q.term(t -> t.field(fullPath).value((Boolean) nestedValue))));
+                                    } else if (nestedValue instanceof String) {
+                                        List<FieldValue> termList = Collections.singletonList(FieldValue.of((String) nestedValue));
+                                        boolQueryBuilder.must(Query.of(q -> q.terms(t -> t.field(fullPath + Constants.KEYWORD)
+                                                .terms(terms -> terms.value(termList))
+                                        )));
+                                    } else if (nestedValue instanceof ArrayList) {
+                                        boolQueryBuilder.must(Query.of(q -> q.terms(t -> t.field(fullPath + Constants.KEYWORD).terms((TermsQueryField) nestedValue))));
+                                    }
+                                });
+                            }
+                        }
+                    });
+            mustNotQueries.forEach(mustNotQuery -> boolQueryBuilder.mustNot(mustNotQuery));
+            boolQueries.forEach(boolQuery -> boolQueryBuilder.must(boolQuery));
+        }
+        return boolQueryBuilder;
+    }
+
+    private void addSortToSearchSourceBuilder(
+            SearchCriteria searchCriteria, SearchRequest.Builder searchRequestBuilder) {
+        if (isNotBlank(searchCriteria.getOrderBy()) && isNotBlank(searchCriteria.getOrderDirection())) {
+            SortOrder sortOrder =
+                    Constants.ASC.equals(searchCriteria.getOrderDirection()) ? SortOrder.Asc : SortOrder.Desc;
+            searchRequestBuilder.sort(SortOptions.of(so -> so
+                    .field(f -> f
+                            .field(searchCriteria.getOrderBy())
+                            .order(sortOrder)
+                    )
+            ));
+        }
+    }
+
+    private void addRequestedFieldsToSearchSourceBuilder(
+            SearchCriteria searchCriteria, SearchRequest.Builder searchRequestBuilder) {
+        if (searchCriteria.getRequestedFields() == null) {
+            // Get all fields in response
+            searchRequestBuilder.source(SourceConfig.of(sc -> sc.fetch(true)));
+        } else {
+            if (searchCriteria.getRequestedFields().isEmpty()) {
+                log.error("Please specify at least one field to include in the results.");
+            }
+            searchRequestBuilder.source(SourceConfig.of(sc -> sc.filter(filter -> filter.includes(searchCriteria.getRequestedFields()))));
+        }
+    }
+
+    private void addQueryStringToFilter(String searchString, BoolQuery.Builder boolQueryBuilder) {
+        if (isNotBlank(searchString)) {
+            Query wildcardQuery = Query.of(q -> q.wildcard(
+                    WildcardQuery.of(w -> w
+                            .field("searchTags.keyword")
+                            .value("*" + searchString.toLowerCase() + "*"))
+            ));
+            boolQueryBuilder.must(wildcardQuery);
+        }
+    }
+
+    private void addFacetsToSearchSourceBuilder(
+            List<String> facets, SearchRequest.Builder searchRequestBuilder) {
+        if (facets != null && !facets.isEmpty()) {
+            Map<String, Aggregation> aggregationMap = facets.stream()
+                    .collect(Collectors.toMap(
+                            field -> field + "_agg",
+                            field -> Aggregation.of(a -> a.terms(
+                                    TermsAggregation.of(t -> t.field(field + ".keyword").size(250))))
+                    ));
+            searchRequestBuilder.aggregations(aggregationMap);
+        }
+    }
+
+    private boolean isNotBlank(String value) {
+        return value != null && !value.trim().isEmpty();
+    }
+
+    @Override
+    public void deleteDocumentsByCriteria(String esIndexName, Query query) {
+        try {
+            HitsMetadata<Object> searchHits = executeSearch(esIndexName, query);
+            assert searchHits.total() != null;
+            if (searchHits.total().value() > 0) {
+                BulkResponse bulkResponse = deleteMatchingDocuments(esIndexName, searchHits);
+                if (!bulkResponse.errors()) {
+                    log.info("Documents matching the criteria deleted successfully from Elasticsearch.");
+                } else {
+                    log.error("Some documents failed to delete from Elasticsearch.");
+                }
+            } else {
+                log.info("No documents match the criteria.");
+            }
+        } catch (Exception e) {
+            log.error("Error occurred during deleting documents by criteria from Elasticsearch.", e);
+        }
+    }
+
+    private HitsMetadata<Object> executeSearch(String esIndexName, Query query) throws IOException {
+        SearchRequest searchRequest = new SearchRequest.Builder()
+                .index(esIndexName)
+                .query(query)
+                .build();
+        SearchResponse<Object> searchResponse =
+                elasticsearchClient.search(searchRequest, Object.class);
+        return searchResponse.hits();
+    }
+
+    private BulkResponse deleteMatchingDocuments(String esIndexName,  HitsMetadata<Object> searchHits)
+            throws IOException {
+        List<BulkOperation> operations = new ArrayList<>();
+        for (Hit<Object> hit : searchHits.hits()) {
+            new DeleteRequest.Builder()
+                    .index(esIndexName)
+                    .id(hit.id())
+                    .build();
+            operations.add(new BulkOperation.Builder().delete(d -> d.index(esIndexName).id(hit.id())).build());
+        }
+        BulkRequest bulkRequest = new BulkRequest.Builder().operations(operations).build();
+        return elasticsearchClient.bulk(bulkRequest);
+    }
+
+    private boolean isRangeQuery(Map<String, Object> nestedMap) {
+        return nestedMap.keySet().stream().anyMatch(key -> key.equals(Constants.SEARCH_OPERATION_GREATER_THAN_EQUALS) ||
+                key.equals(Constants.SEARCH_OPERATION_LESS_THAN_EQUALS) || key.equals(Constants.SEARCH_OPERATION_GREATER_THAN) ||
+                key.equals(Constants.SEARCH_OPERATION_LESS_THAN));
+    }
+
+    private Query buildQueryPart(Map<String, Object> queryMap) {
+        log.info("Search:: buildQueryPart");
+        if (queryMap == null || queryMap.isEmpty()) {
+            return QueryBuilders.matchAll().build()._toQuery();
+        }
+        for (Entry<String, Object> entry : queryMap.entrySet()) {
+            String key = entry.getKey();
+            Object value = entry.getValue();
+
+            switch (key) {
+                case Constants.BOOL:
+                    return buildBoolQuery((Map<String, Object>) value)._toQuery();
+                case Constants.TERM:
+                    return buildTermQuery((Map<String, Object>) value);
+                case Constants.TERMS:
+                    return buildTermsQuery((Map<String, Object>) value);
+                case Constants.MATCH:
+                    return buildMatchQuery((Map<String, Object>) value);
+                case Constants.RANGE:
+                    return buildRangeQuery((Map<String, Object>) value);
+                default:
+                    throw new IllegalArgumentException(Constants.UNSUPPORTED_QUERY + key);
+            }
+        }
+
+        return null;
+    }
+
+    private BoolQuery buildBoolQuery(Map<String, Object> boolMap) {
+        log.info("Search:: builderBoolQuery");
+        BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
+        if (boolMap.containsKey(Constants.MUST)) {
+            List<Map<String, Object>> mustList = (List<Map<String, Object>>) boolMap.get("must");
+            mustList.forEach(must -> boolQueryBuilder.must(buildQueryPart(must)));
+        }
+        if (boolMap.containsKey(Constants.FILTER)) {
+            List<Map<String, Object>> filterList = (List<Map<String, Object>>) boolMap.get("filter");
+            filterList.forEach(filter -> boolQueryBuilder.filter(buildQueryPart(filter)));
+        }
+        if (boolMap.containsKey(Constants.MUST_NOT)) {
+            List<Map<String, Object>> mustNotList = (List<Map<String, Object>>) boolMap.get("must_not");
+            mustNotList.forEach(mustNot -> boolQueryBuilder.mustNot(buildQueryPart(mustNot)));
+        }
+        if (boolMap.containsKey(Constants.SHOULD)) {
+            List<Map<String, Object>> shouldList = (List<Map<String, Object>>) boolMap.get("should");
+            shouldList.forEach(should -> boolQueryBuilder.should(buildQueryPart(should)));
+        }
+
+        return boolQueryBuilder.build();
+    }
+
+    private Query buildTermQuery(Map<String, Object> termMap) {
+        log.info("search::buildTermQuery");
+        BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
+        for (Entry<String, Object> entry : termMap.entrySet()) {
+            boolQueryBuilder.must(QueryBuilders.term(t -> t.field(entry.getKey()).value((FieldValue) entry.getValue())));
+        }
+        return boolQueryBuilder.build()._toQuery();
+    }
+
+    private Query buildTermsQuery(Map<String, Object> termsMap) {
+        log.info("search:: buildTermsQuery");
+        BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
+        for (Entry<String, Object> entry : termsMap.entrySet()) {
+            boolQueryBuilder.must(QueryBuilders.terms(t -> t.field(entry.getKey()).terms((TermsQueryField) entry.getValue())));
+        }
+        return boolQueryBuilder.build()._toQuery();
+    }
+
+    private Query buildMatchQuery(Map<String, Object> matchMap) {
+        log.info("search:: buildMatchQuery");
+        BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
+        for (Entry<String, Object> entry : matchMap.entrySet()) {
+            boolQueryBuilder.must(QueryBuilders.match(m -> m.field(entry.getKey()).query((FieldValue) entry.getValue())));
+        }
+        return boolQueryBuilder.build()._toQuery();
+    }
+
+    private Query buildRangeQuery(Map<String, Object> rangeMap) {
+        log.info("search:: buildRangeQuery");
+        BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
+        for (Entry<String, Object> entry : rangeMap.entrySet()) {
+            Map<String, Object> rangeConditions = (Map<String, Object>) entry.getValue();
+            RangeQuery.Builder rangeQueryBuilder = new RangeQuery.Builder().field(entry.getKey());
+            rangeConditions.forEach((condition, value) -> {
+                switch (condition) {
+                    case "gt":
+                        rangeQueryBuilder.gt(JsonData.of(value));
+                        break;
+                    case "gte":
+                        rangeQueryBuilder.gte(JsonData.of(value));
+                        break;
+                    case "lt":
+                        rangeQueryBuilder.lt(JsonData.of(value));
+                        break;
+                    case "lte":
+                        rangeQueryBuilder.lte(JsonData.of(value));
+                        break;
+                    default:
+                        throw new IllegalArgumentException(Constants.UNSUPPORTED_RANGE + condition);
+                }
+            });
+            boolQueryBuilder.must(rangeQueryBuilder.build()._toQuery());
+        }
+        return boolQueryBuilder.build()._toQuery();
+    }
+
+    @Override
+    public boolean isIndexPresent(String indexName) {
+        try {
+            GetIndexRequest request = new GetIndexRequest.Builder().index(indexName).build();
+            GetIndexResponse response = elasticsearchClient.indices().get(request);
+            return response != null;
+        } catch (IOException e) {
+            log.error("Error checking if index exists", e);
+            return false;
+        }
+    }
+
+    @Override
+    public BulkResponse saveAll(String esIndexName, List<JsonNode> entities) throws IOException {
+        try {
+            log.info("EsUtilServiceImpl :: saveAll");
+            List<BulkOperation> operations = new ArrayList<>();
+            entities.forEach(entity -> {
+                String formattedId = entity.get(Constants.ID).asText();
+                Map<String, Object> entityMap = objectMapper.convertValue(entity, Map.class);
+                BulkOperation operation = BulkOperation.of(b -> b
+                        .index(i -> i
+                                .index(esIndexName)
+                                .id(formattedId)
+                                .document(entityMap)
+                        )
+                );
+                operations.add(operation);
+            });
+
+            BulkRequest bulkRequest = BulkRequest.of(b -> b.operations(operations));
+            return elasticsearchClient.bulk(bulkRequest);
+        } catch (Exception e) {
+            log.error(e.getMessage());
+            throw new CustomException("error bulk uploading", e.getMessage(),
+                HttpStatus.INTERNAL_SERVER_ERROR);
+        }
+    }
+
+
+}
diff --git a/src/main/java/com/igot/cb/transactional/redis/cache/CacheService.java b/src/main/java/com/igot/cb/transactional/redis/cache/CacheService.java
index 4828f50..2132e63 100644
--- a/src/main/java/com/igot/cb/transactional/redis/cache/CacheService.java
+++ b/src/main/java/com/igot/cb/transactional/redis/cache/CacheService.java
@@ -115,4 +115,28 @@ public Map<String, String> getCourseMetadataAsJsonString(List<String> courseIds)
         }
         return result;
     }
+
+    public void removeCache(String key) {
+        try (Jedis jedis = jedisPool.getResource()) {
+            jedis.del(key);
+            logger.debug("Cache key {} removed from redis", key);
+        } catch (Exception e) {
+            logger.error("Error removing cache key {}", key, e);
+        }
+    }
+
+    public List<Object> hget(List<String> keys) {
+        List<Object> resultList = new ArrayList<>();
+        try (Jedis jedis = jedisDataPopulationPool.getResource()) {
+            // Default index is 0, no need to select
+            for (String key : keys) {
+                List<String> result = jedis.hmget(key, key);
+                String value = org.springframework.util.StringUtils.isEmpty(result) ? null : result.get(0);
+                resultList.add(value);
+            }
+        } catch (Exception e) {
+            logger.error("Error in hget: ", e);
+        }
+        return resultList;
+    }
 }
diff --git a/src/main/java/com/igot/cb/transactional/redis/config/RedisConfig.java b/src/main/java/com/igot/cb/transactional/redis/config/RedisConfig.java
index 4d23512..34de748 100644
--- a/src/main/java/com/igot/cb/transactional/redis/config/RedisConfig.java
+++ b/src/main/java/com/igot/cb/transactional/redis/config/RedisConfig.java
@@ -1,10 +1,13 @@
 package com.igot.cb.transactional.redis.config;
 
+import com.igot.cb.transactional.elasticsearch.dto.SearchResult;
 import com.igot.cb.util.CbServerProperties;
+import com.igot.cb.util.Constants;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.RedisTemplate;
 import redis.clients.jedis.JedisPool;
 import redis.clients.jedis.JedisPoolConfig;
 
@@ -44,4 +47,18 @@ private JedisPoolConfig buildPoolConfig() {
         poolConfig.setBlockWhenExhausted(cbProperties.getRedisBlockWhenExhausted());
         return poolConfig;
     }
+
+    @Bean(name = Constants.SEARCH_RESULT_REDIS_TEMPLATE)
+    public RedisTemplate<String, SearchResult> searchResultRedisTemplate() {
+        org.springframework.data.redis.connection.jedis.JedisConnectionFactory jedisConnectionFactory = new org.springframework.data.redis.connection.jedis.JedisConnectionFactory();
+        jedisConnectionFactory.setHostName(cbProperties.getRedisHostName());
+        jedisConnectionFactory.setPort(Integer.parseInt(cbProperties.getRedisPort()));
+        jedisConnectionFactory.afterPropertiesSet();
+        RedisTemplate<String, SearchResult> template = new RedisTemplate<>();
+        template.setConnectionFactory(jedisConnectionFactory);
+        template.setKeySerializer(new org.springframework.data.redis.serializer.StringRedisSerializer());
+        template.setValueSerializer(new org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer<>(SearchResult.class));
+        template.afterPropertiesSet();
+        return template;
+    }
 }
diff --git a/src/main/java/com/igot/cb/util/CbServerProperties.java b/src/main/java/com/igot/cb/util/CbServerProperties.java
index f28af06..1906245 100644
--- a/src/main/java/com/igot/cb/util/CbServerProperties.java
+++ b/src/main/java/com/igot/cb/util/CbServerProperties.java
@@ -191,6 +191,12 @@ public List<String> getProfileCompletionRequiredFields() {
     @Value("${master.data.search.allowed.sortby.fields}")
     private String masterDataAllowedSortByFields;
 
+    @Value("${achievement.status.update.required.fields}")
+    private String requiredFieldsProperty;
+
+    @Value("${elastic.required.field.achievement.json.path}")
+    private String achievementEsRequiredFieldsMappingPath;
+
     public List<String> getMasterDataAllowedSortByFields() {return Arrays.asList(masterDataAllowedSortByFields.split(","));}
 
     public List<String> getMasterDataAllowedType() {
diff --git a/src/main/java/com/igot/cb/util/Constants.java b/src/main/java/com/igot/cb/util/Constants.java
index a18fa1b..a767f8a 100644
--- a/src/main/java/com/igot/cb/util/Constants.java
+++ b/src/main/java/com/igot/cb/util/Constants.java
@@ -469,6 +469,30 @@ public class Constants {
     public static final String INVALID_AUTH_TOKEN = "Invalid user auth token";
     public static final String NAME_KEYWORD = "name.keyword";
     public static final String DESCRIPTION_KEYWORD = "description.keyword";
+    public static final String ACHIEVEMENT_ID = "achievementId";
+    public static final String REJECTED = "REJECTED";
+    public static final String FIELD_CONTEXT_TYPE = "contexttype";
+    public static final String FIELD_APPROVED_BY = "approvedby";
+    public static final String FIELD_APPROVED_ON = "approvedon";
+    public static final String CONTEXT_TYPE_ACHIEVEMENTS = "achievements";
+    public static final String API_ACHIEVEMENT_STATUS_UPDATE = "api.achievement.status.update";
+    public static final int CASSANDRA_FETCH_LIMIT = 1;
+    public static final String LEARNER_ACHIEVEMENT_TABLE ="learner_achievements_v2";
+    public static final String LEARNER_ACHIEVEMENT_INDEX = "achievement_entity";
+    public static final String PENDING = "PENDING";
+    public static final String API_ACHIEVEMENT_SEARCH = "api.achievement.search";
+    public static final String USER_ID_LOWER = "userid";
+    public static final String CONTEXT_TYPE_KEY = "contextType";
+    public static final String SEARCH_RESULT_REDIS_TEMPLATE = "searchResultRedisTemplate";
+    public static final String SEARCH_RESULTS = "search_results";
+    public static final String MINIMUM_CHARACTERS_NEEDED= "Minimum 3 characters are required to search";
+    public static final String USER_PREFIX = "user:" ;
+    public static final String USER_ID_KEY = "user_id";
+    public static final String USER_TABLE = "user";
+    public static final String FIRST_NAME_KEY = "first_name";
+    public static final String FIRST_NAME_CAMEL_CASE = "firstName";
+    public static final String APPROVED_KEY = "APPROVED";
+
     private Constants() {
     }
 }
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
index 8c78df9..324640b 100644
--- a/src/main/resources/application.properties
+++ b/src/main/resources/application.properties
@@ -157,6 +157,9 @@ master.data.search.allowed.sortby.fields=name,description,addedOn,updatedOn
 
 learner.service.host=http://learnerServiceIP:9010
 password.reset.path=/private/user/v1/password/reset
-
-
-
+elasticsearch.host=localhost
+elasticsearch.port=9200
+elasticsearch.username=
+elasticsearch.password=
+achievement.status.update.required.fields=learnerId,id,status,reason,contextType
+elastic.required.field.achievement.json.path =/esFieldsMapping/achievementEsRequiredFieldsMapping.json
\ No newline at end of file
diff --git a/src/main/resources/esFieldsMapping/achievementEsRequiredFieldsMapping.json b/src/main/resources/esFieldsMapping/achievementEsRequiredFieldsMapping.json
new file mode 100644
index 0000000..2fa3d3b
--- /dev/null
+++ b/src/main/resources/esFieldsMapping/achievementEsRequiredFieldsMapping.json
@@ -0,0 +1,45 @@
+{
+  "userId": {
+    "type": "keyword"
+  },
+  "id": {
+    "type": "keyword"
+  },
+  "contextType": {
+    "type": "keyword"
+  },
+  "status": {
+    "type": "keyword"
+  },
+  "orgId": {
+    "type": "keyword"
+  },
+  "approvedBy": {
+    "type": "keyword"
+  },
+  "source": {
+    "type": "text",
+    "fields": {
+      "keyword": {
+        "type": "keyword",
+        "ignore_above": 256
+      }
+    }
+  },
+  "approvedOn": {
+    "type": "date"
+  },
+  "createdOn": {
+    "type": "date"
+  },
+  "contextData": {
+    "type": "object",
+    "dynamic": true
+  },
+  "updatedOn": {
+    "type": "date"
+  },
+  "updatedBy": {
+    "type": "keyword"
+  }
+}
\ No newline at end of file