-
Notifications
You must be signed in to change notification settings - Fork 276
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Rbroughan/retry states api handler (#7484)
- Loading branch information
Showing
11 changed files
with
595 additions
and
3 deletions.
There are no files selected for viewing
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
57 changes: 57 additions & 0 deletions
57
airbyte-server/src/main/java/io/airbyte/server/apis/JobRetryStatesApiController.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,57 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.server.apis; | ||
|
|
||
| import static io.airbyte.commons.auth.AuthRoleConstants.ADMIN; | ||
|
|
||
| import io.airbyte.api.generated.JobRetryStatesApi; | ||
| import io.airbyte.api.model.generated.JobIdRequestBody; | ||
| import io.airbyte.api.model.generated.JobRetryStateRequestBody; | ||
| import io.airbyte.api.model.generated.RetryStateRead; | ||
| import io.airbyte.commons.server.errors.IdNotFoundKnownException; | ||
| import io.airbyte.commons.server.scheduling.AirbyteTaskExecutors; | ||
| import io.airbyte.server.handlers.RetryStatesHandler; | ||
| import io.micronaut.http.HttpStatus; | ||
| import io.micronaut.http.annotation.Controller; | ||
| import io.micronaut.http.annotation.Post; | ||
| import io.micronaut.http.annotation.Status; | ||
| import io.micronaut.scheduling.annotation.ExecuteOn; | ||
| import io.micronaut.security.annotation.Secured; | ||
|
|
||
| @SuppressWarnings("MissingJavadocType") | ||
| @Controller("/api/v1/jobs/retry_states") | ||
| public class JobRetryStatesApiController implements JobRetryStatesApi { | ||
|
|
||
| private final RetryStatesHandler handler; | ||
|
|
||
| @SuppressWarnings("MissingJavadocType") | ||
| public JobRetryStatesApiController(final RetryStatesHandler handler) { | ||
| this.handler = handler; | ||
| } | ||
|
|
||
| @Secured({ADMIN}) | ||
| @ExecuteOn(AirbyteTaskExecutors.IO) | ||
| @Post(uri = "/get") | ||
| @Override | ||
| public RetryStateRead get(final JobIdRequestBody req) { | ||
| final var found = handler.getByJobId(req); | ||
|
|
||
| if (found.isEmpty()) { | ||
| throw new IdNotFoundKnownException(String.format("Could not find Retry State for job_id: %d.", req.getId()), String.valueOf(req.getId())); | ||
| } | ||
|
|
||
| return found.get(); | ||
| } | ||
|
|
||
| @Secured({ADMIN}) | ||
| @ExecuteOn(AirbyteTaskExecutors.IO) | ||
| @Post(uri = "/create_or_update") | ||
| @Override | ||
| @Status(HttpStatus.NO_CONTENT) | ||
| public void createOrUpdate(final JobRetryStateRequestBody req) { | ||
| handler.putByJobId(req); | ||
| } | ||
|
|
||
| } |
42 changes: 42 additions & 0 deletions
42
airbyte-server/src/main/java/io/airbyte/server/handlers/RetryStatesHandler.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,42 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.server.handlers; | ||
|
|
||
| import io.airbyte.api.model.generated.JobIdRequestBody; | ||
| import io.airbyte.api.model.generated.JobRetryStateRequestBody; | ||
| import io.airbyte.api.model.generated.RetryStateRead; | ||
| import io.airbyte.server.handlers.api_domain_mapping.RetryStatesMapper; | ||
| import io.airbyte.server.repositories.RetryStatesRepository; | ||
| import jakarta.inject.Singleton; | ||
| import java.util.Optional; | ||
|
|
||
| /** | ||
| * Interface layer between the API and Persistence layers. | ||
| */ | ||
| @SuppressWarnings("MissingJavadocMethod") | ||
| @Singleton | ||
| public class RetryStatesHandler { | ||
|
|
||
| final RetryStatesRepository repo; | ||
| final RetryStatesMapper mapper; | ||
|
|
||
| public RetryStatesHandler(final RetryStatesRepository repo, final RetryStatesMapper mapper) { | ||
| this.repo = repo; | ||
| this.mapper = mapper; | ||
| } | ||
|
|
||
| public Optional<RetryStateRead> getByJobId(final JobIdRequestBody req) { | ||
| final var found = repo.findByJobId(req.getId()); | ||
|
|
||
| return found.map(mapper::map); | ||
| } | ||
|
|
||
| public void putByJobId(final JobRetryStateRequestBody req) { | ||
| final var model = mapper.map(req); | ||
|
|
||
| repo.createOrUpdateByJobId(model.getJobId(), model); | ||
| } | ||
|
|
||
| } |
45 changes: 45 additions & 0 deletions
45
...server/src/main/java/io/airbyte/server/handlers/api_domain_mapping/RetryStatesMapper.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,45 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.server.handlers.api_domain_mapping; | ||
|
|
||
| import io.airbyte.api.model.generated.JobRetryStateRequestBody; | ||
| import io.airbyte.api.model.generated.RetryStateRead; | ||
| import io.airbyte.server.repositories.domain.RetryState; | ||
| import jakarta.inject.Singleton; | ||
|
|
||
| /** | ||
| * Maps between the API and Persistence layers. It is not static to be injectable and enable easier | ||
| * testing in dependents. | ||
| */ | ||
| @Singleton | ||
| @SuppressWarnings("MissingJavadocMethod") | ||
| public class RetryStatesMapper { | ||
|
|
||
| // API to Domain | ||
| public RetryState map(final JobRetryStateRequestBody api) { | ||
| return RetryState.builder() | ||
| .id(api.getId()) | ||
| .connectionId(api.getConnectionId()) | ||
| .jobId(api.getJobId()) | ||
| .successiveCompleteFailures(api.getSuccessiveCompleteFailures()) | ||
| .totalCompleteFailures(api.getTotalCompleteFailures()) | ||
| .successivePartialFailures(api.getSuccessivePartialFailures()) | ||
| .totalPartialFailures(api.getTotalPartialFailures()) | ||
| .build(); | ||
| } | ||
|
|
||
| // Domain to API | ||
| public RetryStateRead map(final RetryState domain) { | ||
| return new RetryStateRead() | ||
| .id(domain.getId()) | ||
| .connectionId(domain.getConnectionId()) | ||
| .jobId(domain.getJobId()) | ||
| .successiveCompleteFailures(domain.getSuccessiveCompleteFailures()) | ||
| .totalCompleteFailures(domain.getTotalCompleteFailures()) | ||
| .successivePartialFailures(domain.getSuccessivePartialFailures()) | ||
| .totalPartialFailures(domain.getTotalPartialFailures()); | ||
| } | ||
|
|
||
| } |
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
103 changes: 103 additions & 0 deletions
103
airbyte-server/src/test/java/io/airbyte/server/apis/JobRetryStatesApiControllerTest.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,103 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.server.apis; | ||
|
|
||
| import static org.mockito.Mockito.when; | ||
|
|
||
| import io.airbyte.api.model.generated.JobIdRequestBody; | ||
| import io.airbyte.api.model.generated.JobRetryStateRequestBody; | ||
| import io.airbyte.api.model.generated.RetryStateRead; | ||
| import io.airbyte.commons.json.Jsons; | ||
| import io.airbyte.server.handlers.RetryStatesHandler; | ||
| import io.micronaut.context.annotation.Replaces; | ||
| import io.micronaut.context.annotation.Requires; | ||
| import io.micronaut.context.env.Environment; | ||
| import io.micronaut.core.util.StringUtils; | ||
| import io.micronaut.http.HttpRequest; | ||
| import io.micronaut.http.HttpStatus; | ||
| import io.micronaut.test.annotation.MockBean; | ||
| import io.micronaut.test.extensions.junit5.annotation.MicronautTest; | ||
| import java.util.Optional; | ||
| import java.util.UUID; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.mockito.Mock; | ||
| import org.mockito.Mockito; | ||
|
|
||
| @SuppressWarnings({"PMD.JUnitTestsShouldIncludeAssert", "MissingJavadocType"}) | ||
| @MicronautTest | ||
| @Requires(property = "mockito.test.enabled", | ||
| defaultValue = StringUtils.TRUE, | ||
| value = StringUtils.TRUE) | ||
| @Requires(env = {Environment.TEST}) | ||
| class JobRetryStatesApiControllerTest extends BaseControllerTest { | ||
|
|
||
| @Mock | ||
| RetryStatesHandler handler = Mockito.mock(RetryStatesHandler.class); | ||
|
|
||
| @MockBean(RetryStatesHandler.class) | ||
| @Replaces(RetryStatesHandler.class) | ||
| RetryStatesHandler mmStreamStatusesHandler() { | ||
| return handler; | ||
| } | ||
|
|
||
| static String PATH_BASE = "/api/v1/jobs/retry_states"; | ||
| static String PATH_GET = PATH_BASE + "/get"; | ||
| static String PATH_PUT = PATH_BASE + "/create_or_update"; | ||
|
|
||
| @Test | ||
| void getForJobFound() throws Exception { | ||
| when(handler.getByJobId(Mockito.any())) | ||
| .thenReturn(Optional.of(new RetryStateRead())); | ||
|
|
||
| testEndpointStatus( | ||
| HttpRequest.POST( | ||
| PATH_GET, | ||
| Jsons.serialize(Fixtures.jobIdReq())), | ||
| HttpStatus.OK); | ||
| } | ||
|
|
||
| @Test | ||
| void getForJobNotFound() throws Exception { | ||
| when(handler.getByJobId(Mockito.any())) | ||
| .thenReturn(Optional.empty()); | ||
|
|
||
| testErrorEndpointStatus( | ||
| HttpRequest.POST( | ||
| PATH_GET, | ||
| Jsons.serialize(Fixtures.jobIdReq())), | ||
| HttpStatus.NOT_FOUND); | ||
| } | ||
|
|
||
| @Test | ||
| void putForJob() throws Exception { | ||
| testEndpointStatus( | ||
| HttpRequest.POST( | ||
| PATH_PUT, | ||
| Jsons.serialize(Fixtures.retryPutReq())), | ||
| HttpStatus.NO_CONTENT); | ||
| } | ||
|
|
||
| static class Fixtures { | ||
|
|
||
| static long jobId1 = 21891253; | ||
|
|
||
| static JobIdRequestBody jobIdReq() { | ||
| return new JobIdRequestBody().id(jobId1); | ||
| } | ||
|
|
||
| static JobRetryStateRequestBody retryPutReq() { | ||
| return new JobRetryStateRequestBody() | ||
| .id(UUID.randomUUID()) | ||
| .connectionId(UUID.randomUUID()) | ||
| .jobId(jobId1) | ||
| .successiveCompleteFailures(8) | ||
| .totalCompleteFailures(12) | ||
| .successivePartialFailures(4) | ||
| .totalPartialFailures(42); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } |
Oops, something went wrong.