-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
17 changed files
with
648 additions
and
315 deletions.
There are no files selected for viewing
25 changes: 25 additions & 0 deletions
25
mod-source-record-storage-server/src/main/java/org/folio/AuthorityDomainKafkaTopic.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| package org.folio; | ||
|
|
||
|
|
||
| import org.folio.kafka.services.KafkaTopic; | ||
|
|
||
| public enum AuthorityDomainKafkaTopic implements KafkaTopic { | ||
| AUTHORITY("authority"); | ||
|
|
||
| private static final String AUTHORITIES_PREFIX = "authorities"; | ||
| private final String topic; | ||
|
|
||
| AuthorityDomainKafkaTopic(String topic) { | ||
| this.topic = topic; | ||
| } | ||
|
|
||
| @Override | ||
| public String moduleName() { | ||
| return AUTHORITIES_PREFIX; | ||
| } | ||
|
|
||
| @Override | ||
| public String topicName() { | ||
| return topic; | ||
| } | ||
| } |
111 changes: 111 additions & 0 deletions
111
...-record-storage-server/src/main/java/org/folio/consumers/AuthorityDomainKafkaHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| package org.folio.consumers; | ||
|
|
||
| import static org.folio.dao.util.RecordDaoUtil.filterRecordByExternalId; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonCreator; | ||
| import com.fasterxml.jackson.annotation.JsonValue; | ||
| import com.fasterxml.jackson.core.type.TypeReference; | ||
| import io.vertx.core.Future; | ||
| import io.vertx.core.json.Json; | ||
| import io.vertx.core.json.jackson.JacksonCodec; | ||
| import io.vertx.kafka.client.consumer.KafkaConsumerRecord; | ||
| import java.util.Collections; | ||
| import java.util.Map; | ||
| import lombok.extern.log4j.Log4j2; | ||
| import org.folio.dao.util.RecordType; | ||
| import org.folio.kafka.AsyncRecordHandler; | ||
| import org.folio.rest.jooq.enums.RecordState; | ||
| import org.folio.services.RecordService; | ||
| import org.folio.services.util.KafkaUtil; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| @Log4j2 | ||
| @Component | ||
| public class AuthorityDomainKafkaHandler implements AsyncRecordHandler<String, String> { | ||
|
|
||
| public static final String DELETE_DOMAIN_EVENT_TYPE = "DELETE"; | ||
| public static final String DELETE_EVENT_SUB_TYPE_FLD = "deleteEventSubType"; | ||
| public static final String TENANT_FIELD = "tenant"; | ||
| private static final TypeReference<Map<String, Object>> MAP_TYPE_REFERENCE = new TypeReference<>() { }; | ||
| private static final String DOMAIN_EVENT_TYPE_HEADER = "domain-event-type"; | ||
| private final RecordService recordService; | ||
|
|
||
| public AuthorityDomainKafkaHandler(RecordService recordService) { | ||
| this.recordService = recordService; | ||
| } | ||
|
|
||
| @Override | ||
| public Future<String> handle(KafkaConsumerRecord<String, String> consumerRecord) { | ||
| log.trace("handle:: Handling kafka record: '{}'", consumerRecord); | ||
| String authorityId = consumerRecord.key(); | ||
| if (isUnexpectedDomainEvent(consumerRecord)) { | ||
| log.debug("handle:: Expected only DELETE domain type. Skipping authority domain kafka record with ID: '{}'", | ||
| authorityId); | ||
| } | ||
|
|
||
| var recordValue = ((JacksonCodec) Json.CODEC).fromString(consumerRecord.value(), MAP_TYPE_REFERENCE); | ||
| var tenantId = recordValue.get(TENANT_FIELD).toString(); | ||
| var deleteEventSubType = | ||
| AuthorityDeleteEventSubType.fromValue(recordValue.get(DELETE_EVENT_SUB_TYPE_FLD).toString()); | ||
|
|
||
| return switch (deleteEventSubType) { | ||
| case SOFT_DELETE -> performSoftDelete(authorityId, tenantId); | ||
| case HARD_DELETE -> performHardDelete(authorityId, tenantId); | ||
| }; | ||
| } | ||
|
|
||
| private Future<String> performSoftDelete(String authorityId, String tenantId) { | ||
| var condition = filterRecordByExternalId(authorityId); | ||
| return recordService.getRecords(condition, RecordType.MARC_AUTHORITY, Collections.emptyList(), 0, 1, tenantId) | ||
| .compose(recordCollection -> { | ||
| if (recordCollection.getRecords().isEmpty()) { | ||
| log.debug("handle:: No records found for externalId '{}', tenantId '{}'", authorityId, tenantId); | ||
| return Future.succeededFuture(); | ||
| } | ||
| var matchedId = recordCollection.getRecords().get(0).getMatchedId(); | ||
| log.info("handle:: Trying to soft-delete records with: matchedId '{}', externalId '{}', tenantId '{}'", | ||
| matchedId, authorityId, tenantId); | ||
| return recordService.updateRecordsState(matchedId, RecordState.DELETED, RecordType.MARC_AUTHORITY, tenantId); | ||
| }) | ||
| .map(authorityId) | ||
| .onFailure(throwable -> | ||
| log.error("handle:: Failed to soft-delete records with externalId '{}', tenantId '{}'", authorityId, tenantId)); | ||
| } | ||
|
|
||
| private Future<String> performHardDelete(String authorityId, String tenantId) { | ||
| return recordService.deleteRecordsByExternalId(authorityId, tenantId) | ||
| .map(authorityId); | ||
| } | ||
|
|
||
| private static boolean isUnexpectedDomainEvent(KafkaConsumerRecord<String, String> consumerRecord) { | ||
| return !KafkaUtil.headerExists(DOMAIN_EVENT_TYPE_HEADER, DELETE_DOMAIN_EVENT_TYPE, consumerRecord.headers()); | ||
| } | ||
|
|
||
| public enum AuthorityDeleteEventSubType { | ||
|
|
||
| SOFT_DELETE("SOFT_DELETE"), | ||
| HARD_DELETE("HARD_DELETE"); | ||
|
|
||
| private final String value; | ||
|
|
||
| AuthorityDeleteEventSubType(String type) { | ||
| this.value = type; | ||
| } | ||
|
|
||
| @JsonValue | ||
| public String getValue() { | ||
| return value; | ||
| } | ||
|
|
||
| @JsonCreator | ||
| public static AuthorityDeleteEventSubType fromValue(String value) { | ||
| for (AuthorityDeleteEventSubType b : AuthorityDeleteEventSubType.values()) { | ||
| if (b.value.equals(value)) { | ||
| return b; | ||
| } | ||
| } | ||
| throw new IllegalArgumentException("Unexpected value '" + value + "'"); | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.