Added test for new case.

ledger/participant-state/kvutils/src/main/scala/com/daml/ledger/participant/state/kvutils/KeyValueConsumption.scala

@@ -343,9 +343,11 @@ object KeyValueConsumption {
             ),
           )
         )
+
       case _ if deduplicated =>
-        //drop all other duplicate submissions
+        // We only emit updates for duplicate transaction submissions.
         None
+
       case DamlLogEntry.PayloadCase.TRANSACTION_REJECTION_ENTRY if invalidRecordTime =>
         val transactionRejectionEntry = wrappedLogEntry.getTransactionRejectionEntry
         val reason = (timeBounds.tooEarlyUntil, timeBounds.tooLateFrom) match {

ledger/participant-state/kvutils/src/test/suite/scala/com/daml/ledger/participant/state/kvutils/KeyValueConsumptionSpec.scala

@@ -182,6 +182,26 @@ class KeyValueConsumptionSpec extends AnyWordSpec with Matchers {
       runAll(testCases)
     }
 
+    "generate a rejection entry for a transaction if its submission has been deduplicated" in {
+      val inputEntry = {
+        val builder = DamlOutOfTimeBoundsEntry.newBuilder
+        builder.setDuplicateUntil(buildTimestamp(aRecordTimeInstant.plusMillis(1)))
+        builder.setEntry(buildLogEntry(TRANSACTION_REJECTION_ENTRY))
+        builder.build
+      }
+      outOfTimeBoundsEntryToUpdate(aRecordTime, inputEntry) match {
+        case Some(Update.CommandRejected(recordTime, completionInfo, FinalReason(status))) =>
+          recordTime shouldBe aRecordTime
+          completionInfo shouldBe Conversions.parseCompletionInfo(
+            parseInstant(recordTime),
+            someSubmitterInfo,
+          )
+          status.code shouldBe Code.ALREADY_EXISTS.value
+          ()
+        case _ => fail()
+      }
+    }
+
     "generate a rejection entry for a configuration if record time is out of time bounds" in {
       def verifyConfigurationRejection(actual: Option[Update]): Unit = actual match {
         case Some(