fix: duplicate entry exception for correlation_ids table. (#4521) (#4533

)

* fix: duplicate entry exception for correlation_ids table.

* feat: add unit test.

---------

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
(cherry picked from commit 8523343)

Co-authored-by: armory-abedonik <106548537+armory-abedonik@users.noreply.github.com>

orca-core-tck/src/main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/PipelineExecutionRepositoryTck.groovy

@@ -550,6 +550,28 @@ abstract class PipelineExecutionRepositoryTck<T extends ExecutionRepository> ext
     status << ExecutionStatus.values()
   }
 
+  def "should return task ref for currently running pipeline by correlation id"() {
+    given:
+    def execution = pipeline {
+      trigger = new DefaultTrigger("manual", "covfefe")
+    }
+    repository().store(execution)
+    repository().updateStatus(execution.type, execution.id, RUNNING)
+
+    when:
+    def result = repository().retrievePipelineForCorrelationId('covfefe')
+
+    then:
+    result.id == execution.id
+
+    when:
+    repository().updateStatus(execution.type, execution.id, SUCCEEDED)
+    repository().retrievePipelineForCorrelationId('covfefe')
+
+    then:
+    thrown(ExecutionNotFoundException)
+  }
+
   def "should return task ref for currently running orchestration by correlation id"() {
     given:
     def execution = orchestration {

orca-redis/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/jedis/RedisExecutionRepository.java

@@ -695,7 +695,7 @@ public PipelineExecution retrieveByCorrelationId(
   @Override
   public PipelineExecution retrievePipelineForCorrelationId(@Nonnull String correlationId)
       throws ExecutionNotFoundException {
-    String key = format("pipelineCorrelation:%s", correlationId);
+    String key = format("correlation:%s", correlationId);
     return getRedisDelegate(key)
         .withCommandsClient(
             correlationRedis -> {

orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt

@@ -893,18 +893,12 @@ class SqlExecutionRepository(
       }
 
       withPool(poolName) {
-        val exists = ctx.fetchExists(
-          ctx.select()
-            .from("correlation_ids")
-            .where(field("id").eq(execution.trigger.correlationId))
-            .and(executionIdField.eq(execution.id))
-        )
-        if (!exists) {
-          ctx.insertInto(table("correlation_ids"))
-            .columns(field("id"), executionIdField)
-            .values(execution.trigger.correlationId, execution.id)
-            .execute()
-        }
+        ctx.insertInto(table("correlation_ids"))
+          .columns(field("id"), executionIdField)
+          .values(execution.trigger.correlationId, execution.id)
+          .onConflict()
+          .doNothing()
+          .execute()
       }
     }
   }