Add check if the name or title was already updated.

src/main/kotlin/KafkaTopology.kt

@@ -44,7 +44,7 @@ import org.memobase.model.UpdateQuery
 class KafkaTopology(
     private val settings: SettingsLoader,
     translationMappers: TranslationMappers,
-    elasticSearchWrapper: ElasticSearchWrapper
+    private val elasticSearchWrapper: ElasticSearchWrapper
 ) {
     private val log = LogManager.getLogger("SearchDocService")
 
@@ -185,10 +185,17 @@ class KafkaTopology(
         schema
             .map { key, value ->
                 when (value) {
-                    is RecordSetSearchDoc -> KeyValue(
-                        "$key#update",
-                        updateQueryBuilder.updateRecordSetName(value.id, value.name)
-                    )
+                    is RecordSetSearchDoc -> {
+                        // Do not update if the name is already the same.
+                        if (value.name == elasticSearchWrapper.getRecordSetName(value.id)) {
+                            KeyValue(key, null)
+                        } else {
+                            KeyValue(
+                                "$key#update",
+                                updateQueryBuilder.updateRecordSetName(value.id, value.name)
+                            )
+                        }
+                    }
                     else -> KeyValue(key, null)
                 }
             }
@@ -206,10 +213,18 @@ class KafkaTopology(
     private fun recordSetUpdate(stream: KStream<String, Schema>, targetField: String) {
         stream.map { key, value ->
             when (value) {
-                is RecordSetSearchDoc -> KeyValue(
-                    "$key#update",
-                    updateInstitutionContainer(value, targetField)
-                )
+                is RecordSetSearchDoc -> {
+                    val update =
+                        updateInstitutionContainer(value, targetField)
+                    if (update == null) {
+                        KeyValue(key, null)
+                    } else {
+                        KeyValue(
+                            "$key#update",
+                            update
+                        )
+                    }
+                }
                 else -> KeyValue(key, null)
             }
         }
@@ -218,27 +233,63 @@ class KafkaTopology(
             .to(updateTopic)
     }
 
-    private fun updateInstitutionContainer(recordSet: RecordSetSearchDoc, targetField: String): UpdateQuery {
+    private fun updateInstitutionContainer(recordSet: RecordSetSearchDoc, targetField: String): UpdateQuery? {
         return when (targetField) {
-            "institution" -> updateQueryBuilder.updateInstitutionContainers(
-                recordSet.id,
-                targetField,
-                recordSet.institution
-            )
-            "masterInstitution" -> updateQueryBuilder.updateInstitutionContainers(
-                recordSet.id,
-                targetField,
-                recordSet.masterInstitution
-            )
-            "originalInstitution" -> updateQueryBuilder.updateInstitutionContainers(
-                recordSet.id,
-                targetField, recordSet.originalInstitution
-            )
-            "accessInstitution" -> updateQueryBuilder.updateInstitutionContainers(
-                recordSet.id,
-                targetField,
-                recordSet.accessInstitution
-            )
+            "institution" -> {
+                if (recordSet.institution.containsAll(
+                        elasticSearchWrapper.getExtraInstitutionsFromRecordSet(recordSet.id, targetField)
+                    )
+                ) {
+                    null
+                } else {
+                    updateQueryBuilder.updateInstitutionContainers(
+                        recordSet.id,
+                        targetField,
+                        recordSet.institution
+                    )
+                }
+            }
+            "masterInstitution" -> {
+                if (recordSet.masterInstitution.containsAll(
+                        elasticSearchWrapper.getExtraInstitutionsFromRecordSet(recordSet.id, targetField)
+                    )
+                ) {
+                    null
+                } else {
+                    updateQueryBuilder.updateInstitutionContainers(
+                        recordSet.id,
+                        targetField,
+                        recordSet.masterInstitution
+                    )
+                }
+            }
+            "originalInstitution" -> {
+                if (recordSet.originalInstitution.containsAll(
+                        elasticSearchWrapper.getExtraInstitutionsFromRecordSet(recordSet.id, targetField)
+                    )
+                ) {
+                    null
+                } else {
+                    updateQueryBuilder.updateInstitutionContainers(
+                        recordSet.id,
+                        targetField, recordSet.originalInstitution
+                    )
+                }
+            }
+            "accessInstitution" -> {
+                if (recordSet.accessInstitution.containsAll(
+                        elasticSearchWrapper.getExtraInstitutionsFromRecordSet(recordSet.id, targetField)
+                    )
+                ) {
+                    null
+                } else {
+                    updateQueryBuilder.updateInstitutionContainers(
+                        recordSet.id,
+                        targetField,
+                        recordSet.accessInstitution
+                    )
+                }
+            }
             else -> throw Exception("Unknown institution type. Set the wrong constant somewhere...")
         }
     }