diff --git a/activity-aggregator/src/main/scala/org/sunbird/activity/actor/ActivityAggregatorActor.scala b/activity-aggregator/src/main/scala/org/sunbird/activity/actor/ActivityAggregatorActor.scala index e5361f2b3..83cc64487 100644 --- a/activity-aggregator/src/main/scala/org/sunbird/activity/actor/ActivityAggregatorActor.scala +++ b/activity-aggregator/src/main/scala/org/sunbird/activity/actor/ActivityAggregatorActor.scala @@ -271,17 +271,12 @@ class ActivityAggregatorActor @Inject()(implicit val cacheUtil: RedisCacheUtil) } private def updateActivityAggregates(courseAggregations: List[UserEnrolmentAgg], requestContext: RequestContext): Unit = { - logger.info(requestContext, s"updateActivityAggregates: Creating batch update queries for ${courseAggregations.size} aggregations") val aggQueries = courseAggregations.map { agg => activityAggUtil.createActivityAggUpdateMap(agg.activityAgg) }.asJava if (!aggQueries.isEmpty) { - logger.info(requestContext, s"updateActivityAggregates: Executing batch update with ${aggQueries.size()} queries to ${activityAggDBInfo.getTableName}") - cassandraOperation.batchUpdate(activityAggDBInfo.getKeySpace, activityAggDBInfo.getTableName, aggQueries, requestContext) - logger.info(requestContext, s"updateActivityAggregates: Batch update completed successfully") - } else { - logger.warn(requestContext, s"updateActivityAggregates: No queries to execute") + cassandraOperation.batchUpdateWithPutAll(activityAggDBInfo.getKeySpace, activityAggDBInfo.getTableName, aggQueries, requestContext) } } diff --git a/activity-aggregator/src/test/scala/org/sunbird/activity/actor/ActivityAggregatorActorTest.scala b/activity-aggregator/src/test/scala/org/sunbird/activity/actor/ActivityAggregatorActorTest.scala index a5f11e322..552c3582d 100644 --- a/activity-aggregator/src/test/scala/org/sunbird/activity/actor/ActivityAggregatorActorTest.scala +++ b/activity-aggregator/src/test/scala/org/sunbird/activity/actor/ActivityAggregatorActorTest.scala @@ -81,6 +81,11 @@ class ActivityAggregatorActorTest .returning(new Response()) .anyNumberOfTimes() + (cassandraOperation.batchUpdateWithPutAll(_: String, _: String, _: util.List[util.Map[String, util.Map[String, Object]]], _: RequestContext)) + .expects(*, "user_activity_agg", *, *) + .returning(new Response()) + .anyNumberOfTimes() + val actor = system.actorOf(Props(new TestableActivityAggregatorActor(cassandraOperation, redisUtil, deDupUtil, contentSearchUtil, certificateUtil))) val request = createUpdateRequest( @@ -128,6 +133,11 @@ class ActivityAggregatorActorTest .returning(new Response()) .anyNumberOfTimes() + (cassandraOperation.batchUpdateWithPutAll(_: String, _: String, _: util.List[util.Map[String, util.Map[String, Object]]], _: RequestContext)) + .expects(*, "user_activity_agg", *, *) + .returning(new Response()) + .anyNumberOfTimes() + val actor = system.actorOf(Props(new TestableActivityAggregatorActor(cassandraOperation, redisUtil, deDupUtil, contentSearchUtil, certificateUtil))) val request = createUpdateRequest( @@ -194,6 +204,11 @@ class ActivityAggregatorActorTest .returning(new Response()) .anyNumberOfTimes() + (cassandraOperation.batchUpdateWithPutAll(_: String, _: String, _: util.List[util.Map[String, util.Map[String, Object]]], _: RequestContext)) + .expects(*, "user_activity_agg", *, *) + .returning(new Response()) + .anyNumberOfTimes() + val actor = system.actorOf(Props(new TestableActivityAggregatorActor(cassandraOperation, redisUtil, deDupUtil, contentSearchUtil, certificateUtil))) val contentsWithInvalid = new util.ArrayList[util.Map[String, AnyRef]]() @@ -296,6 +311,11 @@ class ActivityAggregatorActorTest .returning(new Response()) .anyNumberOfTimes() + (cassandraOperation.batchUpdateWithPutAll(_: String, _: String, _: util.List[util.Map[String, util.Map[String, Object]]], _: RequestContext)) + .expects(*, "user_activity_agg", *, *) + .returning(new Response()) + .anyNumberOfTimes() + val actor = system.actorOf(Props(new TestableActivityAggregatorActor(cassandraOperation, redisUtil, deDupUtil, contentSearchUtil, certificateUtil))) val inputContents = new util.ArrayList[util.Map[String, AnyRef]]() diff --git a/assessment-aggregator/src/main/scala/org/sunbird/assessment/actor/AssessmentAggregatorActor.scala b/assessment-aggregator/src/main/scala/org/sunbird/assessment/actor/AssessmentAggregatorActor.scala index 3b8732103..a4985b8b8 100644 --- a/assessment-aggregator/src/main/scala/org/sunbird/assessment/actor/AssessmentAggregatorActor.scala +++ b/assessment-aggregator/src/main/scala/org/sunbird/assessment/actor/AssessmentAggregatorActor.scala @@ -1,6 +1,5 @@ package org.sunbird.assessment.actor import org.sunbird.actor.core.BaseActor -import javax.inject.Inject import org.apache.pekko.actor.Props import org.sunbird.common.exception.ProjectCommonException import org.sunbird.common.request.{Request, RequestContext} @@ -12,23 +11,35 @@ import org.sunbird.common.models.util.{JsonKey, LoggerUtil, ProjectUtil} import scala.collection.JavaConverters._ import org.apache.commons.lang3.StringUtils -class AssessmentAggregatorActor @Inject()(_redisService: Option[RedisService],_contentService: Option[ContentService],_cassandraService: Option[CassandraService],_kafkaService: Option[KafkaService]) extends BaseActor { +class AssessmentAggregatorActor( + _cassandraService: Option[CassandraService], + _kafkaService: Option[KafkaService], + _redisService: Option[RedisService], + _contentService: Option[ContentService] +) extends BaseActor { def this() = this(None, None, None, None) - private lazy val redisService = _redisService.getOrElse(new RedisService()) - private lazy val contentService = _contentService.getOrElse(new ContentService()) + private lazy val cassandraService = _cassandraService.getOrElse(AssessmentAggregatorActor.cassandraService) + private lazy val kafkaService = _kafkaService.getOrElse(AssessmentAggregatorActor.kafkaService) + private lazy val redisService = _redisService.getOrElse(AssessmentAggregatorActor.redisService) + private lazy val contentService = _contentService.getOrElse(AssessmentAggregatorActor.contentService) private lazy val assessmentService = new AssessmentService(redisService, contentService) - private lazy val cassandraService = _cassandraService.getOrElse(new CassandraService()) - private lazy val kafkaService = _kafkaService.getOrElse(new KafkaService()) - + override def onReceive(request: Request): Unit = { + request.getOperation match { + case "aggregateAssessment" => aggregateAssessment(request) + case _ => onReceiveUnsupportedOperation(request.getOperation) + } + } + + private def aggregateAssessment(request: Request): Unit = { val replyTo = sender() - try { - processAggregation(request, replyTo) + try { + processAggregation(request, replyTo) } catch { case ex: Exception => - logger.error(request.getRequestContext, "Request failed", ex) + logger.error(request.getRequestContext, s"Assessment aggregation failed: ${ex.getMessage}", ex) replyTo ! createErrorResponse("SERVER_ERROR", ex.getMessage, ResponseCode.SERVER_ERROR.getResponseCode) } } @@ -64,7 +75,7 @@ class AssessmentAggregatorActor @Inject()(_redisService: Option[RedisService],_c replyTo ! createSuccess(assessment.attemptId) } catch { case ex: Exception => - logger.error(context, s"Assessment request failed. Reason: ${ex.getMessage} | Data: $body", ex) + logger.error(context, s"[ASSESSMENT_ACTOR] Request failed: ${ex.getMessage}", ex) replyTo ! createErrorResponse("CLIENT_ERROR", ex.getMessage, ResponseCode.CLIENT_ERROR.getResponseCode) } } @@ -89,7 +100,6 @@ class AssessmentAggregatorActor @Inject()(_redisService: Option[RedisService],_c logger.warn(context, s"Sync Flow: No stored events found for userId=${request.userId}, contentId=${request.contentId}, attemptId=${request.attemptId}") return List(request) } - logger.info(context, s"Sync Flow: Recovered ${existing.size} attempt(s) for userId=${request.userId}, contentId=${request.contentId}") existing.map(toSyncRequest(request, _)) } @@ -125,16 +135,15 @@ class AssessmentAggregatorActor @Inject()(_redisService: Option[RedisService],_c if (skipMissing) { val totalQuestions = metadata.totalQuestions if (totalQuestions > 0 && uniqueEvents.size > totalQuestions) { - logger.warn(context, s"Skipping assessment ${req.attemptId}: unique events (${uniqueEvents.size}) exceed total questions ($totalQuestions)") + logger.warn(context, s"[ASSESSMENT_ACTOR] SKIPPED: unique events (${uniqueEvents.size}) exceed total questions ($totalQuestions) for attemptId=${req.attemptId}") return } } val scoreMetrics = assessmentService.computeScoreMetrics(uniqueEvents) val existing = cassandraService.getAssessment(req.attemptId, req.userId, req.courseId, req.batchId, req.contentId, context) val existingTs = existing.map(_.lastAttemptedOn).getOrElse(0L) - logger.info(context, s"AssessmentAggregatorActor: Comparing timestamps for attemptId=${req.attemptId} | Incoming=${req.assessmentTimestamp} | Existing=$existingTs") if (!req.ignoreTimestampValidation && existingTs > req.assessmentTimestamp) { - logger.info(context, s"Skipping stale assessment: ${req.attemptId}") + logger.warn(context, s"[ASSESSMENT_ACTOR] SKIPPED: Stale assessment attemptId=${req.attemptId}") return } val result = AssessmentResult(req.attemptId, req.userId, req.courseId, req.batchId, req.contentId, scoreMetrics.totalScore, scoreMetrics.totalMaxScore, scoreMetrics.grandTotal, scoreMetrics.questions, existing.map(_.createdOn).getOrElse(System.currentTimeMillis()), req.assessmentTimestamp) @@ -150,7 +159,10 @@ class AssessmentAggregatorActor @Inject()(_redisService: Option[RedisService],_c val attemptId = assessmentService.getLatestAttemptId(agg) if (ProjectUtil.getConfigValue("assessment_aggregator_publish_certificate") == "true") { kafkaService.publishCertificateEvent(userId, courseId, batchId, attemptId) + logger.info(context, s"[ASSESSMENT_ACTOR] Published certificate event for attemptId=$attemptId") } + } else { + logger.warn(context, s"[ASSESSMENT_ACTOR] No assessments found for userId=$userId, courseId=$courseId, batchId=$batchId, contentId=$contentId") } } @@ -225,5 +237,15 @@ class AssessmentAggregatorActor @Inject()(_redisService: Option[RedisService],_c } object AssessmentAggregatorActor { - def props(): Props = Props(new AssessmentAggregatorActor()) + lazy val cassandraService = new CassandraService() + lazy val kafkaService = new KafkaService() + lazy val redisService = new RedisService() + lazy val contentService = new ContentService() + + def props(): Props = Props(new AssessmentAggregatorActor( + Some(cassandraService), + Some(kafkaService), + Some(redisService), + Some(contentService) + )) } diff --git a/assessment-aggregator/src/main/scala/org/sunbird/assessment/service/CassandraService.scala b/assessment-aggregator/src/main/scala/org/sunbird/assessment/service/CassandraService.scala index 58a2bf8f6..27eeb0c8b 100644 --- a/assessment-aggregator/src/main/scala/org/sunbird/assessment/service/CassandraService.scala +++ b/assessment-aggregator/src/main/scala/org/sunbird/assessment/service/CassandraService.scala @@ -48,16 +48,18 @@ class CassandraService(optionalDao: Option[CassandraOperation] = None) { try { if (agg.aggregates.nonEmpty || agg.aggregateDetails.nonEmpty) { val lastUpdated = agg.aggregates.map { case (k, _) => k -> new java.util.Date() } - val data = Map( + val compositeKey = Map( "activity_id" -> cid, "activity_type" -> "Course", "context_id" -> s"cb:$bid", - "user_id" -> uid, + "user_id" -> uid + ).asJava.asInstanceOf[java.util.Map[String, AnyRef]] + val updateAttributes = Map( "aggregates" -> agg.aggregates.asJava, "agg_details" -> agg.aggregateDetails.map(_.toJson).asJava, "agg_last_updated" -> lastUpdated.asJava ).asJava.asInstanceOf[java.util.Map[String, AnyRef]] - dao.upsertRecord(keyspace, activityTable, data, ctx) + dao.updateRecordWithPutAll(keyspace, activityTable, updateAttributes, compositeKey, ctx) } } catch { case e: Exception => logger.error(s"Activity update failed for $uid", e); throw e } } diff --git a/assessment-aggregator/src/test/scala/org/sunbird/assessment/actor/AssessmentAggregatorActorSpec.scala b/assessment-aggregator/src/test/scala/org/sunbird/assessment/actor/AssessmentAggregatorActorSpec.scala index 5c0a8bd1a..483b129f3 100644 --- a/assessment-aggregator/src/test/scala/org/sunbird/assessment/actor/AssessmentAggregatorActorSpec.scala +++ b/assessment-aggregator/src/test/scala/org/sunbird/assessment/actor/AssessmentAggregatorActorSpec.scala @@ -15,7 +15,7 @@ import org.sunbird.common.models.response.Response import java.util.HashMap import scala.collection.JavaConverters._ import scala.concurrent.duration._ -import org.sunbird.assessment.service.{CassandraService, ContentMetadata, ContentService, KafkaService, RedisService} +import org.sunbird.assessment.service.{AssessmentService, CassandraService, ContentMetadata, ContentService, KafkaService, RedisService} import org.sunbird.common.responsecode.ResponseCode class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggregatorActorSpec")) @@ -30,7 +30,9 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre TestKit.shutdownActorSystem(system) } - def getActorRef = TestActorRef(new AssessmentAggregatorActor(Some(mRedis), Some(mContent), Some(mCassandra), Some(mKafka))) + def getActorRef = { + TestActorRef(new AssessmentAggregatorActor(Some(mCassandra), Some(mKafka), Some(mRedis), Some(mContent))) + } "AssessmentAggregatorActor" should "silently ignore unknown message types (standard BaseActor behavior)" in { val actorRef = getActorRef @@ -48,6 +50,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() body.put(JsonKey.USER_ID, "u1") @@ -83,6 +86,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() body.put(JsonKey.USER_ID, "u1") @@ -109,6 +113,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() @@ -140,6 +145,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre reset(mRedis, mContent, mCassandra, mKafka) val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() @@ -178,6 +184,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() body.put(JsonKey.USER_ID, "u1") @@ -196,6 +203,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre reset(mRedis, mContent, mCassandra, mKafka) val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() body.put(JsonKey.COURSE_ID, "c1") @@ -211,13 +219,16 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre val existing = ExistingAssessment("a1", "cont1", System.currentTimeMillis(), System.currentTimeMillis(), 10.0, 10.0, List.empty) when(mRedis.isValidContent(anyString, anyString)).thenReturn(true) when(mRedis.getTotalQuestionsCount(anyString)).thenReturn(Some(10)) - when(mCassandra.getAssessment(anyString, anyString, anyString, anyString, anyString, any[RequestContext])).thenReturn(Some(existing)) - when(mCassandra.getUserAssessments(anyString, anyString, anyString, anyString, any[RequestContext])).thenReturn(List(existing)) + when(mCassandra.getAssessment(anyString, anyString, anyString, anyString, anyString, any[RequestContext])) + .thenReturn(Some(existing)) + when(mCassandra.getUserAssessments(anyString, anyString, anyString, anyString, any[RequestContext])) + .thenReturn(List(existing)) PropertiesCache.getInstance().saveConfigProperty("assessment_aggregator_publish_certificate", "true") val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() body.put(JsonKey.USER_ID, "u1") @@ -239,6 +250,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre reset(mRedis, mContent, mCassandra, mKafka) val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() body.put("userId", "u1") @@ -255,14 +267,31 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre body.put("events", events) request.setRequest(body) - when(mRedis.isValidContent(anyString, anyString)).thenReturn(true) - when(mRedis.getTotalQuestionsCount(anyString)).thenReturn(Some(10)) - when(mCassandra.getAssessment(anyString, anyString, anyString, anyString, anyString, any[RequestContext])) - .thenReturn(Some(ExistingAssessment("att1", "cont1", 2000L, 1000L, 5.0, 10.0, List.empty))) + org.mockito.Mockito.doReturn(true).when(mRedis).isValidContent(org.mockito.ArgumentMatchers.anyString, org.mockito.ArgumentMatchers.anyString) + org.mockito.Mockito.doReturn(Some(10)).when(mRedis).getTotalQuestionsCount(org.mockito.ArgumentMatchers.anyString) + org.mockito.Mockito.doReturn(Some(ExistingAssessment("att1", "cont1", 2000L, 1000L, 5.0, 10.0, List.empty))) + .when(mCassandra).getAssessment( + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.any(classOf[RequestContext]) + ) + + org.mockito.Mockito.doReturn(List(ExistingAssessment("att1", "cont1", 2000L, 1000L, 5.0, 10.0, List.empty))) + .when(mCassandra).getUserAssessments( + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.anyString, + org.mockito.ArgumentMatchers.any(classOf[RequestContext]) + ) + actorRef ! request expectMsgType[Response] - verify(mCassandra, never).saveAssessment(any[AssessmentResult], any[RequestContext]) + verify(mCassandra, never).saveAssessment(org.mockito.ArgumentMatchers.any(classOf[AssessmentResult]), org.mockito.ArgumentMatchers.any(classOf[RequestContext])) } it should "throw exception when content validation fails" in { @@ -270,6 +299,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre PropertiesCache.getInstance().saveConfigProperty("assessment_enable_content_validation", "true") val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() body.put("userId", "u1") @@ -291,6 +321,7 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre // Passing a message that might cause an internal exception (e.g., if a mandatory field is missing in Request) val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequest(null) // This should cause a NPE in processAggregation actorRef ! request expectMsgType[ProjectCommonException].getResponseCode should be (500) @@ -300,11 +331,14 @@ class AssessmentAggregatorActorSpec extends TestKit(ActorSystem("AssessmentAggre reset(mRedis, mContent, mCassandra, mKafka) when(mRedis.isValidContent(any[String], any[String])).thenReturn(true) when(mRedis.getTotalQuestionsCount(any[String])).thenReturn(Some(1)) - when(mCassandra.getAssessment(any[String], any[String], any[String], any[String], any[String], any[RequestContext])).thenReturn(None) - when(mCassandra.getUserAssessments(any[String], any[String], any[String], any[String], any[RequestContext])).thenReturn(List.empty) + when(mCassandra.getAssessment(any[String], any[String], any[String], any[String], any[String], any[RequestContext])) + .thenReturn(None) + when(mCassandra.getUserAssessments(any[String], any[String], any[String], any[String], any[RequestContext])) + .thenReturn(List.empty) val actorRef = getActorRef val request = new Request() + request.setOperation("aggregateAssessment") request.setRequestContext(mock[RequestContext]) val body = new HashMap[String, AnyRef]() body.put("userId", "u1") diff --git a/assessment-aggregator/src/test/scala/org/sunbird/assessment/service/CassandraServiceSpec.scala b/assessment-aggregator/src/test/scala/org/sunbird/assessment/service/CassandraServiceSpec.scala index 9772d2bd8..76cc7fbb6 100644 --- a/assessment-aggregator/src/test/scala/org/sunbird/assessment/service/CassandraServiceSpec.scala +++ b/assessment-aggregator/src/test/scala/org/sunbird/assessment/service/CassandraServiceSpec.scala @@ -37,7 +37,7 @@ class CassandraServiceSpec extends AnyFlatSpec with Matchers with MockitoSugar { val service = new CassandraService(Some(mDao)) val agg = UserActivityAggregate("u1", "c1", "b1", scala.collection.immutable.Map("score:cont1" -> 10.0), List.empty) service.updateUserActivity("u1", "c1", "b1", agg, mock[RequestContext]) - verify(mDao).upsertRecord(anyString, anyString, any[java.util.Map[String, Object]], any[RequestContext]) + verify(mDao).updateRecordWithPutAll(anyString, anyString, any[java.util.Map[String, Object]], any[java.util.Map[String, Object]], any[RequestContext]) } it should "not update user activity if aggregates are empty" in { @@ -45,7 +45,7 @@ class CassandraServiceSpec extends AnyFlatSpec with Matchers with MockitoSugar { val service = new CassandraService(Some(mDao)) val agg = UserActivityAggregate("u1", "c1", "b1", scala.collection.immutable.Map.empty[String, Double], List.empty) service.updateUserActivity("u1", "c1", "b1", agg, mock[RequestContext]) - verify(mDao, never).upsertRecord(anyString, anyString, any, any) + verify(mDao, never).updateRecordWithPutAll(anyString, anyString, any, any, any) } it should "get timestamp from row correctly" in { diff --git a/core/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandra/CassandraOperation.java b/core/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandra/CassandraOperation.java index 4c6f693d3..ee986c92e 100644 --- a/core/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandra/CassandraOperation.java +++ b/core/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandra/CassandraOperation.java @@ -317,6 +317,23 @@ Response updateRecord( Map compositeKey, RequestContext requestContext); + /** + * Updates a record in a Cassandra table using a composite primary key and putAll strategy for map columns. + * + * @param keyspaceName The Cassandra keyspace name. + * @param tableName The table name where the record will be updated. + * @param updateAttributes A map of column names to their updated values. + * @param compositeKey A map representing the composite primary key. + * @param requestContext The request context for tracking and logging. + * @return Response object containing the operation result. + */ + Response updateRecordWithPutAll( + String keyspaceName, + String tableName, + Map updateAttributes, + Map compositeKey, + RequestContext requestContext); + /** * Retrieves a record by its identifier with specified fields. * @@ -528,6 +545,22 @@ Response batchUpdateById( List> records, RequestContext requestContext); + /** + * Performs a batch update operation on multiple records in a Cassandra table. + * For map type columns, it uses 'putAll' behavior (merging) instead of 'set' (replacing). + * + * @param keyspaceName The Cassandra keyspace name. + * @param tableName The table name where records will be updated. + * @param list A list of maps containing the update specifications. + * @param requestContext The request context for tracking and logging. + * @return Response object containing the operation result. + */ + Response batchUpdateWithPutAll( + String keyspaceName, + String tableName, + List>> list, + RequestContext requestContext); + /** * Applies a callback operation on Cassandra async read call. * This method performs asynchronous read operations and applies the provided callback when the diff --git a/core/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandraimpl/CassandraOperationImpl.java b/core/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandraimpl/CassandraOperationImpl.java index aec024c07..31a04b703 100644 --- a/core/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandraimpl/CassandraOperationImpl.java +++ b/core/sunbird-cassandra-utils/src/main/java/org/sunbird/cassandraimpl/CassandraOperationImpl.java @@ -1521,6 +1521,76 @@ requestContext, formatLogMessage("Successfully updated record with composite key return response; } + @Override + public Response updateRecordWithPutAll( + String keyspaceName, + String tableName, + Map updateAttributes, + Map compositeKey, + RequestContext requestContext) { + + long startTime = System.currentTimeMillis(); + + + Response response = new Response(); + Statement updateQuery = null; + + if (updateAttributes == null || updateAttributes.isEmpty()) { + response.put(Constants.RESPONSE, Constants.SUCCESS); + return response; + } + + if (compositeKey == null || compositeKey.isEmpty()) { + throw new ProjectCommonException( + ResponseCode.invalidPropertyError.getErrorCode(), + "Composite key cannot be null or empty for update operation", + ResponseCode.CLIENT_ERROR.getResponseCode()); + } + + try { + Session session = connectionManager.getSession(keyspaceName); + updateQuery = CassandraUtil.createUpdateQueryWithPutAll(compositeKey, updateAttributes, keyspaceName, tableName); + session.execute(updateQuery); + response.put(Constants.RESPONSE, Constants.SUCCESS); + } catch (Exception e) { + if (e.getMessage() != null && e.getMessage().contains(JsonKey.UNKNOWN_IDENTIFIER)) { + String errorMsg = CassandraUtil.processExceptionForUnknownIdentifier(e); + logError( + requestContext, "Invalid column/property error during composite key update with putAll - keyspace: {}, table: {}, error: {}", + keyspaceName, + tableName, + errorMsg, + e); + + throw new ProjectCommonException( + ResponseCode.invalidPropertyError.getErrorCode(), + errorMsg, + ResponseCode.CLIENT_ERROR.getResponseCode()); + } + logError( + requestContext, "Database update operation with putAll failed (composite key) - keyspace: {}, table: {}, error: {}", + keyspaceName, + tableName, + e.getMessage(), + e); + + throw new ProjectCommonException( + ResponseCode.dbUpdateError.getErrorCode(), + ResponseCode.dbUpdateError.getErrorMessage(), + ResponseCode.SERVER_ERROR.getResponseCode()); + + } finally { + if (updateQuery != null) { + logQueryElapseTime( + "updateRecordWithPutAll", startTime, updateQuery.toString(), requestContext); + } else { + logQueryElapseTime("updateRecordWithPutAll", startTime); + } + } + + return response; + } + /** @@ -2564,6 +2634,97 @@ requestContext, formatLogMessage("Successfully batch updated records - keyspace: return response; } + @Override + public Response batchUpdateWithPutAll( + String keyspaceName, + String tableName, + List>> list, + RequestContext requestContext) { + + long startTime = System.currentTimeMillis(); + int recordCount = list != null ? list.size() : 0; + + if (recordCount > 1000) { + logWarn( + requestContext, formatLogMessage("Large batch update detected - keyspace: {}, table: {}, records: {} - Consider splitting into smaller batches for better performance", + keyspaceName, + tableName, + recordCount)); + } + + Response response = new Response(); + BatchStatement batchStatement = new BatchStatement(); + + if (list == null || list.isEmpty()) { + response.put(Constants.RESPONSE, Constants.SUCCESS); + return response; + } + + try { + Session session = connectionManager.getSession(keyspaceName); + for (Map> record : list) { + if (record == null) { + logWarn(requestContext, "Skipping null record in batch update"); + continue; + } + Map primaryKey = record.get(JsonKey.PRIMARY_KEY); + Map nonPKRecord = record.get(JsonKey.NON_PRIMARY_KEY); + if (primaryKey == null || primaryKey.isEmpty()) { + logError( + requestContext, "Invalid record in batch update - missing or empty PRIMARY_KEY for table: {}", + tableName); + throw new ProjectCommonException( + ResponseCode.SERVER_ERROR.getErrorCode(), + "Invalid record structure: PRIMARY_KEY is required", + ResponseCode.SERVER_ERROR.getResponseCode()); + } + + if (nonPKRecord == null || nonPKRecord.isEmpty()) { + logWarn( + requestContext, formatLogMessage("Skipping record with empty NON_PRIMARY_KEY - no fields to update for table: {}", + tableName)); + continue; + } + batchStatement.add( + CassandraUtil.createUpdateQueryWithPutAll(primaryKey, nonPKRecord, keyspaceName, tableName)); + } + if (batchStatement.size() == 0) { + logInfo(requestContext, "No valid records to update in batch"); + response.put(Constants.RESPONSE, Constants.SUCCESS); + return response; + } + ResultSet resultSet = session.execute(batchStatement); + response.put(Constants.RESPONSE, Constants.SUCCESS); + } catch (ProjectCommonException e) { + throw e; + } catch (Exception e) { + logError( + requestContext, "Batch update failed - keyspace: {}, table: {}, records: {}, error: {}", + keyspaceName, + tableName, + recordCount, + e.getMessage(), + e); + + throw new ProjectCommonException( + ResponseCode.SERVER_ERROR.getErrorCode(), + ResponseCode.SERVER_ERROR.getErrorMessage(), + ResponseCode.SERVER_ERROR.getResponseCode()); + } finally { + if (batchStatement != null && batchStatement.size() > 0) { + logQueryElapseTime( + "batchUpdateWithPutAll", + startTime, + batchStatement.getStatements().toString(), + requestContext); + } else { + logQueryElapseTime("batchUpdateWithPutAll", startTime); + } + } + + return response; + } + /** diff --git a/core/sunbird-cassandra-utils/src/main/java/org/sunbird/common/CassandraUtil.java b/core/sunbird-cassandra-utils/src/main/java/org/sunbird/common/CassandraUtil.java index 2592c3866..4b8edcdce 100644 --- a/core/sunbird-cassandra-utils/src/main/java/org/sunbird/common/CassandraUtil.java +++ b/core/sunbird-cassandra-utils/src/main/java/org/sunbird/common/CassandraUtil.java @@ -391,6 +391,46 @@ public static RegularStatement createUpdateQuery( return where; } + /** + * Constructs a Cassandra UPDATE statement using QueryBuilder with putAll for Map types. + * This handles merging of map columns. + * + * @param primaryKey A map of primary key column names to their values (for WHERE clause). + * @param nonPKRecord A map of non-primary key column names to their new values (for SET clause). + * @param keyspaceName The Cassandra keyspace name. + * @param tableName The table name to update. + * @return A RegularStatement representing the UPDATE query. + */ + public static RegularStatement createUpdateQueryWithPutAll( + Map primaryKey, + Map nonPKRecord, + String keyspaceName, + String tableName) { + + Update update = QueryBuilder.update(keyspaceName, tableName); + Assignments assignments = update.with(); + Update.Where where = update.where(); + nonPKRecord + .entrySet() + .stream() + .forEach( + x -> { + if (x.getValue() instanceof Map) { + assignments.and(QueryBuilder.putAll(x.getKey(), (Map) x.getValue())); + } else { + assignments.and(QueryBuilder.set(x.getKey(), x.getValue())); + } + }); + primaryKey + .entrySet() + .stream() + .forEach( + x -> { + where.and(QueryBuilder.eq(x.getKey(), x.getValue())); + }); + return where; + } + /** * Constructs a Cassandra DELETE statement using QueryBuilder. * The statement deletes rows matching the specified primary key.