diff --git a/maas-client/CHANGELOG.md b/maas-client/CHANGELOG.md index 7a1ca539a0..bc18966314 100644 --- a/maas-client/CHANGELOG.md +++ b/maas-client/CHANGELOG.md @@ -1,5 +1,48 @@ # This page contains notably changes of maas-client project. +## Unreleased +* `Features` + - Calls to maas-agent survive a database leader switchover. Retryable: `IOException`, 5xx, 429, + and 405 when the response reason names a database that cannot be written — maas-service reports + a read-only database that way, so the usual "fail fast on 4xx" rule does not hold here. 401 is + not retried, because the token source refreshes on its own schedule. See "Retry behaviour and + configuration" in README. + - New configuration: `maas.http.retry.max-total-duration-ms`, 60s by default, bounding a whole + call including retries. Attempt count and backoff growth derive from it, and each attempt is + capped by what is left, so the worst case a caller sees is that duration rather than the + duration plus one `maas.http.timeout`. `0` disables retries; an unreadable or negative value + warns and falls back to the default. +* `Behaviour changes` + - **A call that fails with a retryable status now takes longer before failing.** It used to throw + on the first unexpected 5xx or 405; it is now retried within the configured duration. + - Failed calls to maas-agent throw `MaaSHttpException` instead of a bare `RuntimeException`. It + extends `MaaSException`, so existing `catch` blocks keep working — note the widening: + `catch (MaaSException)` used to mean a MaaS business error and now also catches transport + failures. + - **`deleteTopic` is not retried, on any options.** It used to be, as any other call. Its response + carries how many topics were deleted, and a repeat of a delete whose response was lost reports + zero for a topic that is already gone. Callers that need the old behaviour must retry themselves + and treat `false` as "not found" rather than "not deleted". + - The Kafka `watch-create` long poll is paced. A down maas-agent used to be re-polled as fast as + the socket could refuse the connection; the poll now backs off exponentially up to 30s with + jitter, so instances that lose the same agent do not all return at the same moment. It keeps + retrying for the life of the client. + - The watch poll window derives from `maas.http.timeout`, 25s with the defaults, instead of a + fixed 60s that outlasted the read timeout and made every quiet poll fail locally. Tuning + `maas.http.timeout` now moves it; below 2s the watch is not usable. + - `KafkaMaaSClient.watchTopicCreate` throws `IllegalStateException` after `close()`, and after the + watch thread has stopped on its own, instead of registering a callback that can never fire. + - Interrupting a thread during a retry wait restores the interrupt flag and aborts, instead of + swallowing `InterruptedException`. +* `Fixed` + - `deleteTopic` threw `NullPointerException` and `search` threw `NoSuchElementException` when + maas-agent answered 200 with an empty body. They now report nothing deleted and no topics found. + - A response without a body no longer throws while the client reads it, in `getOrCreateTopic`, + `deleteTopic` and the error paths. +* `Dependencies` + - New runtime dependency `dev.failsafe:failsafe`, which carries the retry policies. It has no + transitive dependencies, but services with dependency convergence rules will see it appear. + ## 10.0.0 * `Features` - **Breaking:** Removed _MaaSAPIClient.loadConfiguration_ from public API. diff --git a/maas-client/README.md b/maas-client/README.md index 8839492463..c575756c2e 100644 --- a/maas-client/README.md +++ b/maas-client/README.md @@ -58,6 +58,62 @@ MaaSClient client = new MaaSAPIClientImpl(() -> M2MManager.getInstance().getToke ``` +## Retry behaviour and configuration + +Every call to maas-agent is retried before giving up, bounded by a single +setting: the maximum total duration of the call. + +| Property | Default | Meaning | +|---|---|---| +| `maas.http.timeout` | `30` (seconds) | connect/read/write timeout of a **single** attempt | +| `maas.http.retry.max-total-duration-ms` | `60000` | how long one call may take in **total**, retries included. `0` disables retries | + +`max-total-duration-ms` is the only retry knob: the attempt count and the pauses +between attempts are derived from it. The first pause is 1s, each next one +doubles, and the cap is a quarter of the total — with the default 60s that gives +1s, 2s, 4s, 8s, 15s, 15s, roughly six attempts when each attempt fails fast. If +attempts hang instead, fewer of them fit into the same duration. Backoff carries ++/-20% jitter so concurrent callers do not retry in lockstep. + +Each attempt is additionally bounded by what is left of the total duration, so +the worst case a caller sees is that total duration itself rather than the total +duration plus one `maas.http.timeout`. + +The 60s default is meant to outlast a database leader switchover while still +failing fast enough to react to a real outage. + +The watch endpoint (`watch-create`) is excluded: it is a long poll with its own +loop and its own backoff. Its window is derived from `maas.http.timeout` and stays +below it — maas-service holds the request open for the whole window and then answers +with an empty list, which the client has to be able to receive. With the default 30s +timeout the window is 25s. + +`deleteTopic` is excluded as well, on any options: its response carries how many +topics were deleted, and a repeat of a delete whose response was lost reports zero +for a topic that is already gone. `getOrCreateTopic` is retried on any options — +maas-service resolves the classifier before it looks at `onTopicExists`, so a +repeated create returns the registration the first attempt made. + +Which responses are retried: + +| Response | Retried | Why | +|---|---|---| +| `IOException` | yes | connection refused/reset while the agent is being rescheduled | +| 5xx | yes | includes the `500` maas-agent returns when it cannot reach maas-service at all | +| 429 | yes | throttling | +| **405** | **only when the `reason` names a database that cannot be written** | maas-service maps PostgreSQL error `25006` (READ ONLY SQL TRANSACTION) to `405`, so a write against a demoted Patroni node during a switchover arrives as `405`, not as `5xx`. A plain `405` — a route removed on the server, an ingress rejecting the method — is permanent and fails fast | +| 401 | no | `CachingTokenSource` refreshes on its own polling interval, so a retry within the backoff reads the same token, and `M2MInterceptor` has already made its own 401 round trip by then | +| other 4xx | no | permanent client errors, failed on the first attempt | + +The 405 entry is deliberate: the usual "retry 5xx, fail fast on 4xx" rule does not +survive a database leader switchover here. + +The `reason` of the error envelope is what decides, not the error code: every +maas-service error carries the same code, so the envelope alone says nothing. The +match is loose — the reason has to mention a database together with `read-only` +or `not active` — so a reworded message on the server still counts, while a `405` +about a read-only *field* does not. + ## Kafka client usage example All MaaS operations for Kafka is collected in [KafkaMaaSClient](https://github.com/Netcracker/qubership-maas-client/blob/main/client/src/main/java/com/netcracker/cloud/maas/client/api/kafka/KafkaMaaSClient.java). To obtain *new* instance of MaaS Kafka client just call: ```java diff --git a/maas-client/client/pom.xml b/maas-client/client/pom.xml index ecb2628e22..8a3e6e6795 100644 --- a/maas-client/client/pom.xml +++ b/maas-client/client/pom.xml @@ -26,6 +26,10 @@ lombok provided + + dev.failsafe + failsafe + com.squareup.okhttp3 diff --git a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/api/MaaSException.java b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/api/MaaSException.java index 0308e5b769..2989f49a5d 100644 --- a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/api/MaaSException.java +++ b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/api/MaaSException.java @@ -1,7 +1,13 @@ package com.netcracker.cloud.maas.client.api; public class MaaSException extends RuntimeException { + public MaaSException(String format, Object...args) { super(String.format(format, args)); } + + /** For a message that is already built and must not go through String.format. */ + protected MaaSException(String message, Throwable cause) { + super(message, cause); + } } diff --git a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/api/MaaSHttpException.java b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/api/MaaSHttpException.java new file mode 100644 index 0000000000..09c6525fb2 --- /dev/null +++ b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/api/MaaSHttpException.java @@ -0,0 +1,13 @@ +package com.netcracker.cloud.maas.client.api; + +/** A call to maas that did not succeed: an unexpected status code or a transport failure. */ +public class MaaSHttpException extends MaaSException { + + public static MaaSHttpException of(String message) { + return new MaaSHttpException(message, null); + } + + public MaaSHttpException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/Env.java b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/Env.java index 4ef51aeb59..e61cc46231 100644 --- a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/Env.java +++ b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/Env.java @@ -35,6 +35,7 @@ public class Env { public static final String PROP_TENANT_MANAGER_URL = "maas.client.tenant-manager.url"; public static final String PROP_TENANT_MANAGER_RECONNECT_TIMEOUT = "maas.client.tenant-manager.reconnect-timeout"; public static final String PROP_HTTP_TIMEOUT = "maas.http.timeout"; + public static final String PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS = "maas.http.retry.max-total-duration-ms"; public static String apiUrl() { return apiUrl(M2MClient.isK8sM2mEnabled()); @@ -109,6 +110,37 @@ public static Duration httpTimeout() { ); } + static final long DEFAULT_HTTP_RETRY_MAX_TOTAL_DURATION_MS = 60_000L; + + /** + * How long one call may take in total, retries included. Zero leaves a single + * attempt; an unreadable or negative value falls back to the default with a warning. + */ + public static Duration httpRetryMaxTotalDuration() { + return Duration.ofMillis( + stringProperty(PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS) + .map(Env::parseRetryDurationMillis) + .orElse(DEFAULT_HTTP_RETRY_MAX_TOTAL_DURATION_MS) + ); + } + + private static long parseRetryDurationMillis(String raw) { + long millis; + try { + millis = Long.parseLong(raw.trim()); + } catch (NumberFormatException e) { + log.warn("Ignoring '{}={}': not a number of milliseconds, using {}ms", + PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, raw, DEFAULT_HTTP_RETRY_MAX_TOTAL_DURATION_MS); + return DEFAULT_HTTP_RETRY_MAX_TOTAL_DURATION_MS; + } + if (millis < 0) { + log.warn("Ignoring '{}={}': negative, using {}ms", + PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, raw, DEFAULT_HTTP_RETRY_MAX_TOTAL_DURATION_MS); + return DEFAULT_HTTP_RETRY_MAX_TOTAL_DURATION_MS; + } + return millis; + } + public static String url2ws(String url) { return url.replaceAll("^http(s?):", "ws$1:"); } @@ -202,8 +234,11 @@ private static Optional microProfileConfigOptional(String key) { Method getOptionalValue = config.getClass().getMethod("getOptionalValue", String.class, Class.class); return (Optional) getOptionalValue.invoke(config, key, String.class); } catch (ClassNotFoundException e) { + // MicroProfile Config is an optional dependency return Optional.empty(); } catch (Throwable e) { + // Throwable, not Exception: a broken config provider fails class initialisation with + // an Error, and the caller must still fall back to system properties and environment log.trace("MicroProfile Config not available or lookup failed for '{}'", key, e); return Optional.empty(); } diff --git a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/http/HttpExecution.java b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/http/HttpExecution.java index 9b738d6838..2e7eafa3d7 100644 --- a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/http/HttpExecution.java +++ b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/http/HttpExecution.java @@ -4,12 +4,19 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; -import lombok.SneakyThrows; +import com.netcracker.cloud.maas.client.api.MaaSHttpException; +import com.netcracker.cloud.maas.client.impl.Env; +import dev.failsafe.Failsafe; +import dev.failsafe.FailsafeException; +import dev.failsafe.RetryPolicy; +import dev.failsafe.RetryPolicyBuilder; import lombok.extern.slf4j.Slf4j; import okhttp3.*; import java.io.IOException; +import java.time.Duration; import java.util.*; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Predicate; @@ -19,11 +26,11 @@ public class HttpExecution { public static final MediaType JSON = MediaType.parse("application/json; charset=utf-8"); - private final int RETRIES_NUMBER = 30; private final OkHttpClient httpClient; private final Request.Builder req; private final List expectedCodes = new ArrayList<>(); private final Map> errorHandler = new HashMap<>(); + private boolean retryEnabled = true; public static final ObjectMapper MAPPER = new ObjectMapper() .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); @@ -68,6 +75,12 @@ public HttpExecution supressError(int code, Consumer handler) { return this; } + /** Performs a single attempt. For callers that own a retry loop, such as a long poll. */ + public HttpExecution noRetry() { + this.retryEnabled = false; + return this; + } + private Function der(OmnivoreFunction deserializer) { return body -> { try { @@ -94,37 +107,153 @@ public Optional sendAndReceive(OmnivoreFunction responseDeseri return sendAndReceive().map(der(responseDeserializer)); } - @SneakyThrows + /** + * Whether the status is worth another attempt. 405 is here because maas-service reports a + * read-only database that way; 401 is not, because the token source refreshes on its own + * schedule and M2MInterceptor has already retried by the time we see one. + */ + private static boolean isRetryableStatus(int code, String body) { + if (code >= 500 || code == 429) { + return true; + } + return code == 405 && isDatabaseUnavailable(body); + } + + /** Wordings of the two maas-service errors a leader switchover produces, and rewordings. */ + private static final List DATABASE_UNAVAILABLE_MARKERS = + List.of("read-only", "read only", "not in 'active' mode", "not active"); + + /** + * Reads the reason of a maas-service error envelope. The word "database" is required next to + * the marker, so an unrelated 405 that happens to mention read-only data stays permanent. + */ + private static boolean isDatabaseUnavailable(String body) { + if (body == null || body.isEmpty()) { + return false; + } + String reason; + try { + reason = MAPPER.readTree(body).path("reason").asText("").toLowerCase(Locale.ROOT); + } catch (JsonProcessingException e) { + return false; // not a maas-service envelope + } + return reason.contains("database") + && DATABASE_UNAVAILABLE_MARKERS.stream().anyMatch(reason::contains); + } + + /** First backoff pause, and the fraction of the total duration a single pause may reach. */ + private static final Duration BASE_DELAY = Duration.ofSeconds(1); + private static final int MAX_DELAY_FRACTION_OF_TOTAL = 4; + private static final double JITTER = 0.2; + + // Response.body() is nullable in OkHttp; a missing body reads as empty. + private static String bodyAsString(Response response) throws IOException { + ResponseBody body = response.body(); + return body == null ? "" : body.string(); + } + + /** Body of a non-2xx response. Never throws: an unreadable body must not become a retry. */ + private static String errorBodyOrPlaceholder(Response response) { + try { + return bodyAsString(response); + } catch (IOException e) { + log.debug("Could not read error response body", e); + return ""; + } + } + + /** Sends the request, retrying under the policy below until it succeeds or runs out of time. */ private Optional sendAndReceive() { Request compiledReq = req.build(); log.debug("Send request: {}", compiledReq); - int attempt = 0; - while (true) { - try (Response response = httpClient.newCall(compiledReq).execute()) { - // check response codes against acceptable list - log.debug("Received status code: {}, expected codes: {}", response.code(), expectedCodes); + long maxTotalMillis = Env.httpRetryMaxTotalDuration().toMillis(); + if (!retryEnabled || maxTotalMillis <= 0) { + return attemptOnce(compiledReq); + } + try { + return Failsafe.with(retryPolicy(compiledReq, maxTotalMillis)).get(context -> + attempt(compiledReq, maxTotalMillis - context.getElapsedTime().toMillis(), true)); + } catch (RetryableStatus | FailsafeException e) { + Throwable last = e instanceof FailsafeException failsafe ? failsafe.getCause() : null; + if (last instanceof InterruptedException interrupted) { + // Failsafe restores the flag; the call still has to abort rather than retry + Thread.currentThread().interrupt(); + throw new MaaSHttpException("Interrupted while waiting to retry maas-agent request", interrupted); + } + throw new MaaSHttpException("Gave up on " + compiledReq + ": ran out of its " + + maxTotalMillis + "ms total duration.\n\tLast attempt: " + + (last != null ? last : e), last); + } + } - if (errorHandler.containsKey(response.code())) { - errorHandler.get(response.code()).accept(response.body().string()); - return Optional.empty(); - } + private RetryPolicy> retryPolicy(Request compiledReq, long maxTotalMillis) { + Duration maxDelay = Duration.ofMillis(Math.max(1, maxTotalMillis / MAX_DELAY_FRACTION_OF_TOTAL)); + RetryPolicyBuilder> policy = RetryPolicy.builder(); + if (BASE_DELAY.compareTo(maxDelay) < 0) { + policy.withBackoff(BASE_DELAY, maxDelay, 2.0); + } else { + policy.withDelay(maxDelay); // too short for the pause to grow + } + return policy + .handle(IOException.class, RetryableStatus.class) + .withJitter(JITTER) + .withMaxAttempts(-1) + .withMaxDuration(Duration.ofMillis(maxTotalMillis)) + .onRetry(event -> log.warn("Retrying request: {}. Attempt {} failed with {}, within {}ms total", + compiledReq, event.getAttemptCount(), event.getLastException(), maxTotalMillis)) + .build(); + } - if (!expectedCodes.contains(response.code())) { - throw new RuntimeException("Unexpected status code " + response.code() + " for request: " + compiledReq + "\n\tResponse body: " + response.body().string()); - } + /** One request/response exchange. Throws {@link RetryableStatus} for an outcome worth repeating. */ + private Optional attempt(Request compiledReq, long remainingMs, boolean retrying) throws IOException { + Call call = httpClient.newCall(compiledReq); + if (retrying) { + call.timeout().timeout(Math.max(1, remainingMs), TimeUnit.MILLISECONDS); + } + try (Response response = call.execute()) { + // check response codes against acceptable list + log.debug("Received status code: {}, expected codes: {}", response.code(), expectedCodes); - String body = response.body().string(); - log.debug("Response body: {}", body); - return Optional.of(body); - } catch (IOException e) { - if (attempt++ < RETRIES_NUMBER) { - log.warn("Error execute http request: {}, Retry {} of {}", e.getMessage(), attempt, RETRIES_NUMBER); - Thread.sleep(1000); - } else { - throw new RuntimeException("Error executing " + compiledReq + ". Number of " + RETRIES_NUMBER + " retries exceeded", e); + if (errorHandler.containsKey(response.code())) { + errorHandler.get(response.code()).accept(bodyAsString(response)); + return Optional.empty(); + } + + if (!expectedCodes.contains(response.code())) { + String errorBody = errorBodyOrPlaceholder(response); + if (retrying && isRetryableStatus(response.code(), errorBody)) { + throw new RetryableStatus(response.code(), errorBody); } + throw MaaSHttpException.of("Unexpected status code " + response.code() + + " for request: " + compiledReq + "\n\tResponse body: " + errorBody); } + + String body = bodyAsString(response); + log.debug("Response body: {}", body); + return Optional.of(body); + } + } + + /** The {@link #noRetry()} path, and a total duration configured to zero. */ + private Optional attemptOnce(Request compiledReq) { + try { + return attempt(compiledReq, 0, false); + } catch (IOException e) { + throw new MaaSHttpException("Error executing " + compiledReq, e); + } + } + + /** A status the caller did not expect, but one worth another attempt. Never leaves this class. */ + private static final class RetryableStatus extends RuntimeException { + + RetryableStatus(int code, String body) { + super("status " + code + ", body: " + body, null, false, false); + } + + @Override + public String toString() { + return getMessage(); } } } diff --git a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImpl.java b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImpl.java index 4e9d0e62f8..f13e2ed7d5 100644 --- a/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImpl.java +++ b/maas-client/client/src/main/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImpl.java @@ -8,9 +8,11 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.function.Consumer; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -23,6 +25,7 @@ import com.netcracker.cloud.maas.client.api.kafka.TopicAddress; import com.netcracker.cloud.maas.client.api.kafka.TopicCreateOptions; import com.netcracker.cloud.maas.client.impl.ApiUrlProvider; +import com.netcracker.cloud.maas.client.impl.Env; import com.netcracker.cloud.maas.client.impl.Lazy; import com.netcracker.cloud.maas.client.impl.dto.kafka.v1.TopicDeleteRequest; import com.netcracker.cloud.maas.client.impl.dto.kafka.v1.TopicDeleteResponse; @@ -32,6 +35,9 @@ import com.netcracker.cloud.maas.client.impl.http.HttpClient; import com.netcracker.cloud.tenantmanager.client.TenantManagerConnector; +import dev.failsafe.Failsafe; +import dev.failsafe.FailsafeException; +import dev.failsafe.RetryPolicy; import lombok.extern.slf4j.Slf4j; @Slf4j @@ -40,10 +46,46 @@ public class KafkaMaaSClientImpl implements KafkaMaaSClient { private final Lazy tenantManagerConnector; private final ApiUrlProvider apiProvider; - private final Duration watchTimeout = Duration.ofSeconds(60); + /** + * How long maas-service is asked to hold a watch poll open. Stays below the read timeout, + * otherwise a quiet poll dies locally instead of returning an empty 200. + */ + private final Duration watchTimeout = watchTimeout(Env.httpTimeout()); + + /** Largest gap left between the watch window and the read timeout. */ + private static final long MAX_WATCH_MARGIN_SECONDS = 5; + + /** + * The margin is clamped, so a small read timeout narrows the window instead of inverting it. + * The window travels in whole seconds, so {@code maas.http.timeout} below 2s is unsupported. + */ + static Duration watchTimeout(Duration httpTimeout) { + long timeoutSeconds = httpTimeout.getSeconds(); + long marginSeconds = Math.min(MAX_WATCH_MARGIN_SECONDS, timeoutSeconds / 2); + return Duration.ofSeconds(Math.max(1, timeoutSeconds - marginSeconds)); + } + + private static final Duration WATCH_RETRY_INTERVAL = Duration.ofSeconds(1); + private static final Duration WATCH_MAX_RETRY_INTERVAL = Duration.ofSeconds(30); + private static final double WATCH_BACKOFF_MULTIPLIER = 2.0; + + /** Keeps instances that lost the same agent from returning to it at the same moment. */ + private static final double WATCH_BACKOFF_JITTER = 0.2; // there is no need in highly concurrent map/lists implementation, we will wait for network responses most of the time private final Map>> topicCreateListeners = Collections.synchronizedMap(new HashMap<>()); private volatile boolean closed = false; + + /** + * Set when the watch thread exits without {@link #close()}. Nothing restarts it, so a later + * {@link #watchTopicCreate} must refuse rather than register a callback that cannot fire. + */ + private volatile boolean watchThreadDead = false; + + /** + * Monitor for parking the watch thread. Not the thread itself: {@link Thread#join()} waits on + * that monitor too, and would steal the notification meant for the loop. + */ + private final Object watchLock = new Object(); private final Lazy watchThread = new Lazy<>(() -> { Thread exec = new Thread(this::watchTenantCreateTopics, "watchTopicCreate"); exec.setDaemon(true); @@ -69,6 +111,8 @@ public TopicAddress getOrCreateTopic(Classifier classifier, TopicCreateOptions o String url = apiProvider.getKafkaTopicUrl(options.getOnTopicExists()); log.info("Get or create topic by classifier=`{}' and options=`{}'", classifier, options); + // Retried on any options: maas-service resolves the classifier before it looks at + // onTopicExists, so a repeat returns the registration the first attempt made. return httpClient.request(url) .post(TopicRequest.builder(classifier).build().options(options)) .expect(HTTP_OK, HTTP_CREATED) @@ -82,15 +126,20 @@ public Optional getTopic(Classifier classifier) { return Optional.ofNullable(searchTopic(classifier)); } + /** Not retried: a repeat reports zero deleted for a topic the first attempt already removed. */ @Override public boolean deleteTopic(Classifier classifier) { TopicDeleteResponse resp = httpClient.request(apiProvider.getKafkaTopicUrl(null)) .delete(new TopicDeleteRequest(classifier)) .expect(HTTP_OK) + .noRetry() .sendAndReceive(TopicDeleteResponse.class) .orElse(null); - if (resp != null && !resp.getFailedToDelete().isEmpty()) { + if (resp == null) { + return false; // empty body + } + if (!resp.getFailedToDelete().isEmpty()) { throw new MaaSException("Error delete topic by classifier: %s. Error: %s", classifier, resp.getFailedToDelete().get(0).getMessage()); } @@ -110,64 +159,160 @@ public void watchTenantTopics(String name, Consumer> callback } private void watchTenantCreateTopics() { - TypeReference> typeRef = new TypeReference<>() { - }; - while (!closed) { - while (!closed && !topicCreateListeners.isEmpty()) { - String url = apiProvider.getKafkaTopicWatchCreateUrl(watchTimeout); - List found = Collections.emptyList(); - try { - found = httpClient.request(url) - .post(topicCreateListeners.keySet()) - .expect(200) - .sendAndReceive(typeRef) - .orElse(Collections.emptyList()); - } catch (Exception e) { - log.error("Error execute request to {}", url, e); + try { + while (!closed) { + if (!pollWhileThereIsSomethingToWatch()) { + return; } - - for (TopicInfo addr : found) { - List> callbacks = topicCreateListeners.remove(addr.getClassifier()); - if (callbacks == null) { - // this is unexpected situation in theory, but with this, code will be a little safer - continue; - } - - for (Consumer callback : callbacks) { - try { - log.info("Topic create event for {} received, execute callback {}", addr.getClassifier(), callback); - callback.accept(new TopicAddressImpl(addr)); - } catch (Exception e) { - log.error("Error execute callback {}", callback, e); - } - } + if (closed || !parkUntilThereIsSomethingToWatch()) { + return; } } - - if (closed) { - return; + } finally { + if (!closed) { + // exiting without close(): let watchTopicCreate refuse further registrations + watchThreadDead = true; } + } + } + /** + * Polls the watch endpoint until nothing is being watched any more. + * + * @return false if the thread must stop + */ + private boolean pollWhileThereIsSomethingToWatch() { + while (!closed && !topicCreateListeners.isEmpty()) { + String url = apiProvider.getKafkaTopicWatchCreateUrl(watchTimeout); + List found; try { - log.info("Nothing to watch, sleep thread."); - synchronized (watchThread.get()) { - watchThread.get().wait(); + found = Failsafe.with(watchRetryPolicy(url)).get(() -> poll(url)); + } catch (Exception e) { + reportWatchStopped(url, e); + return false; + } + deliver(found); + } + return true; + } + + /** Reports why the watch thread is stopping, unless it is a normal {@link #close()}. */ + private void reportWatchStopped(String url, Exception e) { + Throwable cause = e instanceof FailsafeException failsafe ? failsafe.getCause() : e; + if (cause instanceof InterruptedException) { + // Failsafe clears the flag when it catches this, so the check below cannot see it + Thread.currentThread().interrupt(); + } + // `closed` is checked too: an interrupt can be swallowed further down + if (closed) { + return; // shutting down, not a failure worth reporting + } + if (Thread.currentThread().isInterrupted()) { + log.error("Watch thread interrupted without close(). Topic create callbacks for {} " + + "will no longer fire; recreate the client to resume watching", watchedNames(), cause); + return; + } + // the policy only gives up on interrupt, so this should be unreachable + log.error("Watch poll for {} stopped unexpectedly. Topic create callbacks for {} " + + "will no longer fire; recreate the client to resume watching", url, watchedNames(), cause); + } + + /** + * Backoff between failed watch polls. Unbounded in attempts, because the subscription lives as + * long as the client; a fresh policy per poll resets the sequence after every success, and the + * interrupt from {@link #close()} ends the wait. + */ + private RetryPolicy> watchRetryPolicy(String url) { + return RetryPolicy.>builder() + .handle(Exception.class) + .withBackoff(WATCH_RETRY_INTERVAL, WATCH_MAX_RETRY_INTERVAL, WATCH_BACKOFF_MULTIPLIER) + .withJitter(WATCH_BACKOFF_JITTER) + .withMaxAttempts(-1) + .onRetry(event -> log.warn("Error execute request to {}. Attempt {} failed with {}, will back off before retrying", + url, event.getAttemptCount(), event.getLastException())) + .build(); + } + + private static final TypeReference> TOPIC_LIST = new TypeReference<>() { + }; + + private Set watchedClassifiers() { + synchronized (topicCreateListeners) { + return new HashSet<>(topicCreateListeners.keySet()); + } + } + + /** Names only: a classifier is an open map and may carry whatever the caller put in it. */ + private List watchedNames() { + return watchedClassifiers().stream().map(Classifier::getName).toList(); + } + + /** One long poll for topics created since the previous call. */ + private List poll(String url) { + return httpClient.request(url) + .post(watchedClassifiers()) + .expect(200) + .noRetry() + .sendAndReceive(TOPIC_LIST) + .orElse(Collections.emptyList()); + } + + /** Hands each created topic to the callbacks registered for it, removing them as it goes. */ + private void deliver(List found) { + for (TopicInfo addr : found) { + List> callbacks = topicCreateListeners.remove(addr.getClassifier()); + if (callbacks == null) { + // this is unexpected situation in theory, but with this, code will be a little safer + continue; + } + for (Consumer callback : callbacks) { + try { + log.info("Topic create event for {} received, execute callback {}", addr.getClassifier(), callback); + callback.accept(new TopicAddressImpl(addr)); + } catch (Exception e) { + log.error("Error execute callback {}", callback, e); } - log.info("Woke up!"); - } catch (InterruptedException e) { - return; // exit loop } } } + /** + * Parks the thread while no topic is being watched. + * + * @return false if the thread was interrupted and must stop + */ + private boolean parkUntilThereIsSomethingToWatch() { + try { + log.info("Nothing to watch, sleep thread."); + synchronized (watchLock) { + while (!closed && topicCreateListeners.isEmpty()) { + watchLock.wait(); + } + } + log.info("Woke up!"); + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + @Override public void watchTopicCreate(String name, Consumer callback) { + if (closed) { + throw new IllegalStateException("Client is closed, cannot watch topic: " + name); + } + if (watchThreadDead) { + throw new IllegalStateException("Watch thread has stopped unexpectedly, cannot watch topic: " + + name + "; recreate the client to resume watching"); + } apiProvider.getServerApiVersion().requiresApiVersion(2, 8); log.info("Add watch for topic by: {}, callback: {}", name, callback); topicCreateListeners.computeIfAbsent(new Classifier(name), k -> Collections.synchronizedList(new ArrayList<>())).add(callback); - synchronized (watchThread.get()) { - watchThread.get().notify(); + watchThread.get(); // start the thread if this is the first watch + synchronized (watchLock) { + watchLock.notifyAll(); } } @@ -220,7 +365,7 @@ public List search(SearchCriteria criteria) { .post(criteria) .expect(HTTP_OK) .sendAndReceive(typeRef) - .get() + .orElseGet(Collections::emptyList) .stream() .map(TopicAddressImpl::new) .collect(Collectors.toList()); @@ -229,8 +374,11 @@ public List search(SearchCriteria criteria) { @Override public void close() { closed = true; + synchronized (watchLock) { + watchLock.notifyAll(); // release the watch thread if it is parked + } if (watchThread.isInitialized()) { - watchThread.get().interrupt(); + watchThread.get().interrupt(); // also cuts a backoff sleep short try { watchThread.get().join(1000); } catch (InterruptedException e) { diff --git a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/EnvTest.java b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/EnvTest.java index 08a85415b1..32d3085710 100644 --- a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/EnvTest.java +++ b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/EnvTest.java @@ -156,4 +156,29 @@ void testMicroserviceName() throws Exception { .execute(Env::microserviceName); assertEquals("abc", value); } + + @Test + void testHttpRetryMaxTotalDurationDefault() { + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, null, () -> + assertEquals(Env.DEFAULT_HTTP_RETRY_MAX_TOTAL_DURATION_MS, + Env.httpRetryMaxTotalDuration().toMillis())); + } + + @Test + void testHttpRetryMaxTotalDurationIsRead() { + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, " 1500 ", () -> + assertEquals(1500, Env.httpRetryMaxTotalDuration().toMillis(), "surrounding spaces are tolerated")); + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "0", () -> + assertEquals(0, Env.httpRetryMaxTotalDuration().toMillis(), "zero disables retries")); + } + + /** An unusable value must not fail the call that happens to be first; it falls back and warns. */ + @Test + void testHttpRetryMaxTotalDurationFallsBackOnUnusableValue() { + for (String raw : new String[]{"soon", "", "-1"}) { + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, raw, () -> + assertEquals(Env.DEFAULT_HTTP_RETRY_MAX_TOTAL_DURATION_MS, + Env.httpRetryMaxTotalDuration().toMillis(), "unusable value: '" + raw + "'")); + } + } } diff --git a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/http/HttpExecutionFailoverTest.java b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/http/HttpExecutionFailoverTest.java new file mode 100644 index 0000000000..e4d93fde49 --- /dev/null +++ b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/http/HttpExecutionFailoverTest.java @@ -0,0 +1,324 @@ +package com.netcracker.cloud.maas.client.impl.http; + +import com.netcracker.cloud.maas.client.api.MaaSHttpException; +import com.netcracker.cloud.maas.client.impl.Env; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockserver.integration.ClientAndServer; +import org.mockserver.junit.jupiter.MockServerExtension; +import org.mockserver.matchers.Times; +import org.mockserver.verify.VerificationTimes; + +import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; + +import static com.netcracker.cloud.maas.client.Utils.withProp; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.params.provider.Arguments.arguments; +import static org.mockserver.model.HttpRequest.request; +import static org.mockserver.model.HttpResponse.response; + +@ExtendWith(MockServerExtension.class) +class HttpExecutionFailoverTest { + + private static final String PATH = "/api/v1/kafka/topic"; + + static Stream retryableResponses() { + return Stream.of( + arguments("a read-only database", 405, + "{\"code\":\"MAAS-0600\",\"reason\":\"database is in read-only mode\"}", 2), + arguments("a database out of 'active' mode", 405, + "{\"code\":\"MAAS-0600\",\"reason\":\"database is not in 'active' mode\"}", 2), + // the reason is matched loosely, so a reworded one still counts + arguments("a reworded read-only database", 405, + "{\"reason\":\"Database is read only\"}", 1), + arguments("maas-agent unable to reach maas-service", 500, + "{\"error\":\"error proxying request: connection refused\"}", 2), + arguments("throttling", 429, "{\"error\":\"slow down\"}", 1) + ); + } + + @ParameterizedTest(name = "{0} is retried") + @MethodSource("retryableResponses") + void testFailover_RetryableResponseSucceedsOnRetry(String description, int status, String body, int failures, + ClientAndServer mockServer) { + mockServer.reset(); + mockServer.when(request().withPath(PATH), Times.exactly(failures)) + .respond(response().withStatusCode(status).withBody(body)); + mockServer.when(request().withPath(PATH), Times.unlimited()) + .respond(response().withStatusCode(200).withBody("\"ok\"")); + + withFastRetries(() -> + assertEquals("ok", execution(mockServer).expect(200).sendAndReceive(String.class).orElseThrow())); + + mockServer.verify(request().withPath(PATH), VerificationTimes.exactly(failures + 1)); + } + + /** Responses that are permanent, so the call must fail on its first attempt. */ + static Stream permanentResponses() { + return Stream.of( + arguments("a plain client error", 400, "{\"error\":\"bad request\"}"), + // the token source refreshes on its own schedule and M2MInterceptor has already + // retried, so repeating the call here only sends the same token again + arguments("a rejected token", 401, "{\"error\":\"unauthorized\"}"), + // 405 is transient only for a read-only database; a route removed on the server + // or an ingress rejecting the method is not + arguments("405 without a maas-service envelope", 405, "Method Not Allowed"), + // every maas-service error carries the same code, so the envelope alone means + // nothing: the reason has to name a database that cannot be written + arguments("405 whose maas-service reason is unrelated", 405, + "{\"code\":\"MAAS-0600\",\"reason\":\"topic 'active-orders' is inactive\"}"), + arguments("405 about a read-only field rather than the database", 405, + "{\"reason\":\"the read-only field cannot be updated\"}"), + arguments("405 with the marker outside the reason", 405, + "{\"message\":\"database is in read-only mode\"}") + ); + } + + @ParameterizedTest(name = "{0} is not retried") + @MethodSource("permanentResponses") + void testFailover_PermanentResponseNotRetried(String description, int status, String body, + ClientAndServer mockServer) { + mockServer.reset(); + mockServer.when(request().withPath(PATH), Times.unlimited()) + .respond(response().withStatusCode(status).withBody(body)); + + withFastRetries(() -> assertMessageContains(String.valueOf(status), execution(mockServer).expect(200))); + + mockServer.verify(request().withPath(PATH), VerificationTimes.exactly(1)); + } + + /** + * Each attempt is clamped to what is left of the total duration, so a hanging + * agent cannot stretch the call past it. + */ + @Test + void testMaxTotalDuration_BoundsAHangingAttempt() throws IOException { + // accepts the connection and never answers, unlike a refused connect which fails fast + try (ServerSocket silentServer = new ServerSocket(0)) { + startAcceptor(silentServer, socket -> { /* hold the connection open and stay silent */ }); + + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "1000", () -> { + OkHttpClient client = new OkHttpClient.Builder() + .readTimeout(Duration.ofMinutes(1)) + .build(); + Request.Builder req = new Request.Builder() + .url("http://127.0.0.1:" + silentServer.getLocalPort() + PATH) + .get(); + HttpExecution execution = new HttpExecution(client, req).expect(200); + + long start = System.currentTimeMillis(); + assertThrows(MaaSHttpException.class, () -> execution.sendAndReceive(String.class)); + long elapsedMs = System.currentTimeMillis() - start; + assertTrue(elapsedMs < 5_000, + "expected the call to be bounded by its 1000ms total duration rather than by the " + + "one minute read timeout, took " + elapsedMs + "ms"); + }); + } + } + + /** The watch long poll owns its own loop, so its execution must send the request exactly once. */ + @Test + void testNoRetry_SendsExactlyOneAttempt(ClientAndServer mockServer) { + mockServer.reset(); + mockServer.when(request().withPath(PATH), Times.unlimited()) + .respond(response().withStatusCode(500).withBody("{\"error\":\"agent down\"}")); + + withFastRetries(() -> { + HttpExecution execution = execution(mockServer).expect(200).noRetry(); + assertThrows(MaaSHttpException.class, () -> execution.sendAndReceive(String.class)); + }); + + mockServer.verify(request().withPath(PATH), VerificationTimes.exactly(1)); + } + + /** + * A zero total duration is the config-level off switch: one attempt, no retries, and no + * per-attempt clamp that would cut that attempt short. + */ + @Test + void testZeroTotalDuration_SendsOneAttemptAndDoesNotRetry(ClientAndServer mockServer) { + mockServer.reset(); + mockServer.when(request().withPath(PATH), Times.unlimited()) + .respond(response().withStatusCode(500).withBody("{\"error\":\"agent down\"}")); + + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "0", () -> { + HttpExecution execution = execution(mockServer).expect(200); + assertThrows(MaaSHttpException.class, () -> execution.sendAndReceive(String.class)); + }); + + mockServer.verify(request().withPath(PATH), VerificationTimes.exactly(1)); + } + + @Test + void testInterrupt_RestoresFlagAndAbortsRetryLoop() throws IOException { + // A server that drops every connection: the attempt fails at once and the loop moves + // into its backoff wait, which is where the interrupt has to land. + try (ServerSocket rudeServer = new ServerSocket(0)) { + CountDownLatch firstAttemptFailed = new CountDownLatch(1); + startAcceptor(rudeServer, socket -> { + socket.close(); + firstAttemptFailed.countDown(); + }); + + // A long total duration keeps the retry wait long enough for the interrupt to land in it. + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "600000", () -> { + Request.Builder req = new Request.Builder() + .url("http://127.0.0.1:" + rudeServer.getLocalPort() + PATH) + .get(); + HttpExecution execution = new HttpExecution(new OkHttpClient(), req).expect(200); + + AtomicBoolean interruptedAfter = new AtomicBoolean(); + AtomicReference thrown = new AtomicReference<>(); + + Thread worker = new Thread(() -> { + try { + execution.sendAndReceive(String.class); + } catch (Exception e) { + thrown.set(e); + } finally { + interruptedAfter.set(Thread.currentThread().isInterrupted()); + } + }, "http-execution-interrupt-test"); + worker.start(); + + assertTrue(firstAttemptFailed.await(10, TimeUnit.SECONDS), "the first attempt never reached the server"); + awaitState(worker, Thread.State.TIMED_WAITING); + worker.interrupt(); + worker.join(10_000); + + assertFalse(worker.isAlive(), "worker should abort instead of continuing to retry after interrupt"); + assertTrue(interruptedAfter.get(), "interrupt flag must be restored after an interrupted retry wait"); + assertInstanceOf(MaaSHttpException.class, thrown.get(), + "the interrupt must surface as a maas exception, not as an unrelated failure"); + assertTrue(thrown.get().getMessage().contains("Interrupted while waiting to retry"), + "unexpected message: " + thrown.get().getMessage()); + }); + } + } + + /** + * Running out of time is the usual way a failover call ends, so the exception has to say what + * kept failing. Without the cause the trace shows only that a minute went by. + */ + @Test + void testTotalDurationExceeded_CarriesTheLastFailureAsCause() throws IOException { + try (ServerSocket rudeServer = new ServerSocket(0)) { + AtomicInteger attempts = new AtomicInteger(); + startAcceptor(rudeServer, socket -> { + attempts.incrementAndGet(); + socket.close(); + }); + + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "1500", () -> { + Request.Builder req = new Request.Builder() + .url("http://127.0.0.1:" + rudeServer.getLocalPort() + PATH) + .get(); + HttpExecution execution = new HttpExecution(new OkHttpClient(), req).expect(200); + + MaaSHttpException e = assertThrows(MaaSHttpException.class, + () -> execution.sendAndReceive(String.class)); + assertTrue(e.getMessage().contains("ran out of its"), "unexpected message: " + e.getMessage()); + assertInstanceOf(IOException.class, e.getCause(), + "the transport failure that consumed the time must be the cause"); + // without this the test would pass on a single attempt that never retried + assertTrue(attempts.get() > 1, + "the time must have been spent on retries, but only " + attempts.get() + " attempt was made"); + }); + } + } + + // A tight max total duration must cut the retry loop short well before the attempt count is exhausted. + @Test + void testMaxTotalDuration_AbortsBeforeAttemptsExhausted(ClientAndServer mockServer) { + mockServer.reset(); + mockServer.when(request().withPath(PATH), Times.unlimited()) + .respond(response().withStatusCode(500).withBody("{\"error\":\"agent down\"}")); + + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "200", () -> { + HttpExecution execution = execution(mockServer).expect(200); + assertThrows(MaaSHttpException.class, () -> execution.sendAndReceive(String.class)); + mockServer.verify(request().withPath(PATH), VerificationTimes.atMost(10)); + }); + } + + // A short total duration is now the only lever: it bounds both the number of attempts + // and the pauses between them (the cap is derived as a quarter of it). + private static void withFastRetries(Runnable test) { + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "5000", test::run); + } + + private static HttpExecution execution(ClientAndServer mockServer) { + OkHttpClient client = new OkHttpClient(); + Request.Builder req = new Request.Builder() + .url("http://localhost:" + mockServer.getPort() + PATH) + .get(); + return new HttpExecution(client, req); + } + + private static void assertMessageContains(String expected, HttpExecution execution) { + MaaSHttpException e = assertThrows(MaaSHttpException.class, () -> execution.sendAndReceive(String.class)); + assertTrue(e.getMessage().contains(expected), "unexpected message: " + e.getMessage()); + } + + @FunctionalInterface + private interface SocketHandler { + void handle(Socket socket) throws IOException; + } + + /** Waits until the thread reaches the given state, so an interrupt lands where the test means it to. */ + private static void awaitState(Thread thread, Thread.State state) { + long deadline = System.currentTimeMillis() + 10_000; + while (thread.getState() != state && System.currentTimeMillis() < deadline) { + Thread.onSpinWait(); + } + assertEquals(state, thread.getState(), "the worker never reached " + state); + } + + /** Serves the socket on a daemon thread until it is closed, then releases what it accepted. */ + private static void startAcceptor(ServerSocket server, SocketHandler handler) { + Thread acceptor = new Thread(() -> { + List accepted = new ArrayList<>(); + try { + while (!server.isClosed()) { + Socket socket = server.accept(); + accepted.add(socket); + handler.handle(socket); + } + } catch (IOException e) { + // the server socket was closed, the test is over + } finally { + accepted.forEach(HttpExecutionFailoverTest::closeQuietly); + } + }, "test-acceptor-" + server.getLocalPort()); + acceptor.setDaemon(true); + acceptor.start(); + } + + private static void closeQuietly(Socket socket) { + try { + socket.close(); + } catch (IOException e) { + // nothing useful to do while tearing a test down + } + } +} diff --git a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImplTest.java b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImplTest.java index f831130e17..61b6144bdc 100644 --- a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImplTest.java +++ b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientImplTest.java @@ -21,6 +21,7 @@ import org.mockserver.matchers.Times; import org.mockserver.mock.action.ExpectationResponseCallback; import org.mockserver.model.HttpRequest; +import org.mockserver.verify.VerificationTimes; import com.netcracker.cloud.maas.client.Utils; import com.netcracker.cloud.maas.client.api.Classifier; @@ -43,6 +44,41 @@ @Slf4j // TODO tests with kafka SSL+password class KafkaMaaSClientImplTest { + /** maas-agent's answer for the `orders' topic, shared by the tests that just need a valid one. */ + private static final String ORDERS_TOPIC_RESPONSE = """ + { + "addresses": { + "PLAINTEXT": [ + "my-kafka.kafka-cluster:9092" + ] + }, + "name": "maas.core_dev.orders.1234567", + "classifier": { + "name": "orders", + "namespace": "core-dev", + "tenantId": "d047619f-6886-4842-81a7-3f87cb748ac1" + }, + "namespace": "core-dev", + "instance": "default", + "requestedSettings": { + "numPartitions": 1, + "replicationFactor": 1, + "replicaAssignment": null, + "configs": null + }, + "actualSettings": { + "numPartitions": 1, + "replicationFactor": 1, + "replicaAssignment": { + "0": [ 0 ] + }, + "configs": { + "cleanup.policy": "delete" + } + } + } + """; + @BeforeEach public void setup(ClientAndServer mockServer) { mockServer.reset(); @@ -102,37 +138,7 @@ public void testGetTenantTopic(ClientAndServer mockServer) { ).respond( response() .withStatusCode(200) - .withBody("{\n" + - " \"addresses\": {\n" + - " \"PLAINTEXT\": [\n" + - " \"my-kafka.kafka-cluster:9092\"\n" + - " ]\n" + - " }, \n" + - " \"name\": \"maas.core_dev.orders.1234567\",\n" + - " \"classifier\": {\n" + - " \"name\": \"orders\",\n" + - " \"namespace\": \"core-dev\",\n" + - " \"tenantId\": \"d047619f-6886-4842-81a7-3f87cb748ac1\"\n" + - " }, \n" + - " \"namespace\": \"core-dev\",\n" + - " \"instance\": \"default\",\n" + - " \"requestedSettings\": {\n" + - " \"numPartitions\": 1,\n" + - " \"replicationFactor\": 1,\n" + - " \"replicaAssignment\": null,\n" + - " \"configs\": null\n" + - " },\n" + - " \"actualSettings\": {\n" + - " \"numPartitions\": 1,\n" + - " \"replicationFactor\": 1,\n" + - " \"replicaAssignment\": {\n" + - " \"0\": [ 0 ]\n" + - " },\n" + - " \"configs\": {\n" + - " \"cleanup.policy\": \"delete\"\n" + - " }\n" + - " } \n" + - "}\n") + .withBody(ORDERS_TOPIC_RESPONSE) ); // run test @@ -245,37 +251,7 @@ public void testNoNullValuesInCache(ClientAndServer mockServer) { ).respond( response() .withStatusCode(200) - .withBody("{\n" + - " \"addresses\": {\n" + - " \"PLAINTEXT\": [\n" + - " \"my-kafka.kafka-cluster:9092\"\n" + - " ]\n" + - " }, \n" + - " \"name\": \"maas.core_dev.orders.1234567\",\n" + - " \"classifier\": {\n" + - " \"name\": \"orders\",\n" + - " \"namespace\": \"core-dev\",\n" + - " \"tenantId\": \"d047619f-6886-4842-81a7-3f87cb748ac1\"\n" + - " }, \n" + - " \"namespace\": \"core-dev\",\n" + - " \"instance\": \"default\",\n" + - " \"requestedSettings\": {\n" + - " \"numPartitions\": 1,\n" + - " \"replicationFactor\": 1,\n" + - " \"replicaAssignment\": null,\n" + - " \"configs\": null\n" + - " },\n" + - " \"actualSettings\": {\n" + - " \"numPartitions\": 1,\n" + - " \"replicationFactor\": 1,\n" + - " \"replicaAssignment\": {\n" + - " \"0\": [ 0 ]\n" + - " },\n" + - " \"configs\": {\n" + - " \"cleanup.policy\": \"delete\"\n" + - " }\n" + - " } \n" + - "}\n") + .withBody(ORDERS_TOPIC_RESPONSE) ); topicAddress = client.getTopic(new Classifier("orders").tenantId("d047619f-6886-4842-81a7-3f87cb748ac1")); @@ -303,37 +279,7 @@ public void testGetOrCreateLazyTenantTopic(ClientAndServer mockServer) { ).respond( response() .withStatusCode(200) - .withBody("{\n" + - " \"addresses\": {\n" + - " \"PLAINTEXT\": [\n" + - " \"my-kafka.kafka-cluster:9092\"\n" + - " ]\n" + - " }, \n" + - " \"name\": \"maas.core_dev.orders.1234567\",\n" + - " \"classifier\": {\n" + - " \"name\": \"orders\",\n" + - " \"namespace\": \"core-dev\",\n" + - " \"tenantId\": \"d047619f-6886-4842-81a7-3f87cb748ac1\"\n" + - " }, \n" + - " \"namespace\": \"core-dev\",\n" + - " \"instance\": \"default\",\n" + - " \"requestedSettings\": {\n" + - " \"numPartitions\": 1,\n" + - " \"replicationFactor\": 1,\n" + - " \"replicaAssignment\": null,\n" + - " \"configs\": null\n" + - " },\n" + - " \"actualSettings\": {\n" + - " \"numPartitions\": 1,\n" + - " \"replicationFactor\": 1,\n" + - " \"replicaAssignment\": {\n" + - " \"0\": [ 0 ]\n" + - " },\n" + - " \"configs\": {\n" + - " \"cleanup.policy\": \"delete\"\n" + - " }\n" + - " } \n" + - "}\n") + .withBody(ORDERS_TOPIC_RESPONSE) ); // run test @@ -450,12 +396,11 @@ public void testGetOrCreateTopicWithRetry(ClientAndServer mockServer) throws IOE "}\n") ); - // run test + // default options: maas-service resolves the classifier before it looks at + // onTopicExists, so the retry after the timed-out first attempt gets the same topic var client = createKafkaClient("http://localhost:" + mockServer.getPort()); TopicAddress topicAddress = client.getOrCreateTopic(new Classifier("orders"), - TopicCreateOptions.builder() - .name("user-test1") - .build()); + TopicCreateOptions.builder().name("user-test1").build()); assertEquals("user-test1", topicAddress.getTopicName()); }); }); @@ -596,6 +541,66 @@ void testTopicDeleteSuccess(ClientAndServer mockServer) throws Exception { }); } + /** An empty 200 used to reach the response fields and throw NPE instead of answering "nothing deleted". */ + @Test + void testTopicDeleteEmptyBody(ClientAndServer mockServer) { + withProp(Env.PROP_NAMESPACE, "cloud-dev", () -> { + withProp(Env.PROP_MAAS_AGENT_URL, "http://localhost:" + mockServer.getPort(), () -> { + + mockServer.when( + request().withMethod("DELETE").withPath("/api/v2/kafka/topic"), Times.once() + ).respond(response().withBody("")); + + KafkaMaaSClient kafkaClient = new MaaSAPIClientImpl(() -> "faketoken", null, null).getKafkaClient(); + assertFalse(kafkaClient.deleteTopic(new Classifier("orders"))); + }); + }); + } + + /** + * A retry after the server completed a delete comes back with empty lists, which the client + * reads as "nothing was deleted". + */ + @Test + void testTopicDeleteIsNotRetried(ClientAndServer mockServer) { + withProp(Env.PROP_NAMESPACE, "cloud-dev", () -> { + withProp(Env.PROP_MAAS_AGENT_URL, "http://localhost:" + mockServer.getPort(), () -> { + + mockServer.when(request().withMethod("DELETE").withPath("/api/v2/kafka/topic"), Times.unlimited()) + .respond(response().withStatusCode(500).withBody("{\"error\":\"agent down\"}")); + + KafkaMaaSClient kafkaClient = new MaaSAPIClientImpl(() -> "faketoken", null, null).getKafkaClient(); + Classifier orders = new Classifier("orders"); + assertThrows(MaaSException.class, () -> kafkaClient.deleteTopic(orders)); + + mockServer.verify(request().withMethod("DELETE").withPath("/api/v2/kafka/topic"), + VerificationTimes.exactly(1)); + }); + }); + } + + /** Create is the operation a switchover interrupts most often, and it retries on any options. */ + @Test + void testGetOrCreateTopicIsRetriedOnDefaultOptions(ClientAndServer mockServer) { + withProp(Env.PROP_NAMESPACE, "cloud-dev", () -> { + withProp(Env.PROP_MAAS_AGENT_URL, "http://localhost:" + mockServer.getPort(), () -> { + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "1000", () -> { + + mockServer.when(request().withMethod("POST").withPath("/api/v2/kafka/topic"), Times.unlimited()) + .respond(response().withStatusCode(500).withBody("{\"error\":\"agent down\"}")); + + KafkaMaaSClient kafkaClient = new MaaSAPIClientImpl(() -> "faketoken", null, null).getKafkaClient(); + Classifier orders = new Classifier("orders"); + assertThrows(MaaSException.class, + () -> kafkaClient.getOrCreateTopic(orders, TopicCreateOptions.DEFAULTS)); + + mockServer.verify(request().withMethod("POST").withPath("/api/v2/kafka/topic"), + VerificationTimes.atLeast(2)); + }); + }); + }); + } + @Test void testTopicDeleteError(ClientAndServer mockServer) throws Exception { withProp(Env.PROP_NAMESPACE, "cloud-dev", () -> { @@ -775,6 +780,16 @@ void testClose(ClientAndServer mockServer) { }); } + @Test + void testWatchTopicCreateThrowsAfterClose(ClientAndServer mockServer) { + withProp(Env.PROP_NAMESPACE, "cloud-dev", () -> { + KafkaMaaSClientImpl client = createKafkaClient("http://localhost:" + mockServer.getPort()); + client.close(); + + assertThrows(IllegalStateException.class, () -> client.watchTopicCreate("orders", addr -> {})); + }); + } + private KafkaMaaSClientImpl createKafkaClient(String agentUrl) { System.setProperty(Env.PROP_MAAS_AGENT_URL, agentUrl); var httpClient = HttpClient.getMaasClient(() -> "faketoken"); diff --git a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientWatchBackoffTest.java b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientWatchBackoffTest.java new file mode 100644 index 0000000000..9f0d1572fb --- /dev/null +++ b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/kafka/KafkaMaaSClientWatchBackoffTest.java @@ -0,0 +1,135 @@ +package com.netcracker.cloud.maas.client.impl.kafka; + +import static com.netcracker.cloud.maas.client.Utils.withProp; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.time.Duration; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.netcracker.cloud.maas.client.impl.ApiUrlProvider; +import com.netcracker.cloud.maas.client.impl.Env; +import com.netcracker.cloud.maas.client.impl.apiversion.ServerApiVersion; +import com.netcracker.cloud.maas.client.impl.http.HttpClient; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; + +/** + * Pins the invariant that a failing {@code watch-create} long poll is retried with a backoff + * rather than in a hot loop. + * + *

Without one, a maas-agent that is down gets polled as fast as the socket can refuse the + * connection — hammering it exactly while it is coming back up. + */ +class KafkaMaaSClientWatchBackoffTest { + + private static final String WATCHED_TOPIC = "orders"; + private static final String NAMESPACE = "cloud-dev"; + + /** Three polls are enough to see the pause between them grow. */ + private static final int OBSERVED_POLLS = 3; + + private final List pollMillis = Collections.synchronizedList(new ArrayList<>()); + private final List pollQueries = Collections.synchronizedList(new ArrayList<>()); + private final CountDownLatch pollsObserved = new CountDownLatch(OBSERVED_POLLS); + private HttpServer agentStub; + private KafkaMaaSClientImpl client; + + @BeforeEach + void startAgentStub() throws IOException { + agentStub = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + agentStub.createContext("/api-version", exchange -> respond(exchange, 200, "{\"major\": 2, \"minor\": 8}")); + agentStub.createContext("/api/v2/kafka/topic/watch-create", this::failWatchPoll); + agentStub.start(); + } + + @AfterEach + void stopClientAndStub() { + if (client != null) { + client.close(); + } + agentStub.stop(0); + } + + /** + * A window that outlasts the read timeout means the empty 200 ending a quiet poll never + * arrives, so every such poll fails locally and walks the backoff up to its cap. + */ + @Test + void watchWindowFormulaStaysBelowTheReadTimeout() { + for (long readTimeoutSeconds : new long[]{2, 5, 6, 10, 30, 60, 120}) { + Duration readTimeout = Duration.ofSeconds(readTimeoutSeconds); + Duration window = KafkaMaaSClientImpl.watchTimeout(readTimeout); + assertTrue(window.compareTo(readTimeout) < 0, + "a " + readTimeoutSeconds + "s read timeout must leave room for the answer, got " + window); + assertFalse(window.isZero() || window.isNegative(), + "the window must stay positive, got " + window); + } + } + + @Test + void failingWatchPollIsBackedOffInsteadOfHotLooping() { + withProp(Env.PROP_NAMESPACE, NAMESPACE, () -> { + String agentUrl = "http://localhost:" + agentStub.getAddress().getPort(); + withProp(Env.PROP_MAAS_AGENT_URL, agentUrl, () -> { + client = createKafkaClient(agentUrl); + client.watchTopicCreate(WATCHED_TOPIC, addr -> { /* never created in this test */ }); + + assertTrue(pollsObserved.await(30, TimeUnit.SECONDS), + "the watch loop reached the agent stub only " + pollMillis.size() + + " times out of " + OBSERVED_POLLS + ", so nothing was measured"); + + for (int poll = 1; poll < OBSERVED_POLLS; poll++) { + long pause = pollMillis.get(poll) - pollMillis.get(poll - 1); + assertTrue(pause > 500, + "expected the watch loop to pause after a failure, but poll " + poll + + " followed the previous one in " + pause + "ms"); + } + + assertEquals("timeout=25s", pollQueries.get(0), + "the poll must carry the window derived from maas.http.timeout"); + }); + }); + } + + private static KafkaMaaSClientImpl createKafkaClient(String agentUrl) { + var httpClient = HttpClient.getMaasClient(() -> "faketoken"); + var serverApiVersion = new ServerApiVersion(httpClient, agentUrl); + + return new KafkaMaaSClientImpl(httpClient, + () -> { throw new UnsupportedOperationException("tenant manager is not used in this test"); }, + new ApiUrlProvider(serverApiVersion, agentUrl)); + } + + /** Answers every poll with 500, the code maas-agent returns when it cannot reach maas-service. */ + private void failWatchPoll(HttpExchange exchange) throws IOException { + pollMillis.add(System.currentTimeMillis()); + pollQueries.add(exchange.getRequestURI().getQuery()); + pollsObserved.countDown(); + respond(exchange, 500, "{\"error\":\"error proxying request: maas-service unavailable\"}"); + } + + private static void respond(HttpExchange exchange, int status, String body) throws IOException { + exchange.getRequestBody().readAllBytes(); + + byte[] payload = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(status, payload.length); + try (OutputStream response = exchange.getResponseBody()) { + response.write(payload); + } + } +} diff --git a/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/rabbit/RabbitFailoverTest.java b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/rabbit/RabbitFailoverTest.java new file mode 100644 index 0000000000..6dcbbe2cb4 --- /dev/null +++ b/maas-client/client/src/test/java/com/netcracker/cloud/maas/client/impl/rabbit/RabbitFailoverTest.java @@ -0,0 +1,106 @@ +package com.netcracker.cloud.maas.client.impl.rabbit; + +import com.netcracker.cloud.maas.client.api.Classifier; +import com.netcracker.cloud.maas.client.api.MaaSHttpException; +import com.netcracker.cloud.maas.client.impl.ApiUrlProvider; +import com.netcracker.cloud.maas.client.impl.Env; +import com.netcracker.cloud.maas.client.impl.apiversion.ServerApiVersion; +import com.netcracker.cloud.maas.client.impl.http.HttpClient; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.mockserver.integration.ClientAndServer; +import org.mockserver.junit.jupiter.MockServerExtension; +import org.mockserver.matchers.Times; +import org.mockserver.verify.VerificationTimes; + +import static com.netcracker.cloud.maas.client.Utils.withProp; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockserver.model.HttpRequest.request; +import static org.mockserver.model.HttpResponse.response; + +@ExtendWith(MockServerExtension.class) +class RabbitFailoverTest { + + private static final String PATH = "/api/v2/rabbit/vhost"; + + private String savedAgentUrl; + + @BeforeEach + void reset(ClientAndServer mockServer) { + savedAgentUrl = System.getProperty(Env.PROP_MAAS_AGENT_URL); + mockServer.reset(); + mockServer.when(request().withPath("/api-version")) + .respond(response().withBody("{\"major\":2, \"minor\": 16}")); + } + + // the agent url points at a mock server port that is gone once this class is done, + // so it must not leak into the rest of the JVM + @AfterEach + void restoreAgentUrl() { + if (savedAgentUrl == null) { + System.clearProperty(Env.PROP_MAAS_AGENT_URL); + } else { + System.setProperty(Env.PROP_MAAS_AGENT_URL, savedAgentUrl); + } + } + + @ParameterizedTest(name = "{0} is retried on the vhost path") + @CsvSource(delimiter = '|', textBlock = """ + a read-only database | 405 | {"code":"MAAS-0600","reason":"database is in read-only mode"} + an unreachable agent | 500 | {"error":"error proxying request: connection refused"} + """) + void testFailover_RetryableResponseSucceedsOnRetry(String description, int status, String body, + ClientAndServer mockServer) { + mockServer.when(request().withMethod("POST").withPath(PATH), Times.exactly(2)) + .respond(response().withStatusCode(status).withBody(body)); + mockServer.when(request().withMethod("POST").withPath(PATH), Times.unlimited()) + .respond(response().withStatusCode(200).withBody(""" + { + "cnn": "ampq://rabbit-cluster:4321/maas.core-dev.123456", + "username": "testuser", + "password": "plain:testpassword" + } + """)); + + withProp(Env.PROP_NAMESPACE, "core-dev", () -> + withFastRetries(() -> { + RabbitMaaSClientImpl client = createRabbitClient("http://localhost:" + mockServer.getPort()); + assertNotNull(client.getOrCreateVirtualHost(new Classifier("commands"))); + })); + + mockServer.verify(request().withMethod("POST").withPath(PATH), VerificationTimes.exactly(3)); + } + + @Test + void testFailover_400NotRetried(ClientAndServer mockServer) { + mockServer.when(request().withMethod("POST").withPath(PATH), Times.unlimited()) + .respond(response().withStatusCode(400).withBody("{\"error\":\"bad request\"}")); + + withProp(Env.PROP_NAMESPACE, "core-dev", () -> + withFastRetries(() -> { + RabbitMaaSClientImpl client = createRabbitClient("http://localhost:" + mockServer.getPort()); + Classifier classifier = new Classifier("commands"); + assertThrows(MaaSHttpException.class, () -> client.getOrCreateVirtualHost(classifier)); + })); + + mockServer.verify(request().withMethod("POST").withPath(PATH), VerificationTimes.exactly(1)); + } + + // A short total duration is now the only lever: it bounds both the number of attempts + // and the pauses between them (the cap is derived as a quarter of it). + private static void withFastRetries(Runnable test) { + withProp(Env.PROP_HTTP_RETRY_MAX_TOTAL_DURATION_MS, "5000", test::run); + } + + private static RabbitMaaSClientImpl createRabbitClient(String agentUrl) { + System.setProperty(Env.PROP_MAAS_AGENT_URL, agentUrl); + var httpClient = HttpClient.getMaasClient(() -> "faketoken"); + var serverApiVersion = new ServerApiVersion(httpClient, agentUrl); + return new RabbitMaaSClientImpl(httpClient, new ApiUrlProvider(serverApiVersion, agentUrl)); + } +} diff --git a/maas-client/pom.xml b/maas-client/pom.xml index c847670231..8473452884 100644 --- a/maas-client/pom.xml +++ b/maas-client/pom.xml @@ -25,6 +25,7 @@ 1.1.3-SNAPSHOT 4.2.1 + 3.3.2 ${project.basedir}/../report-aggregate/target/site/jacoco-aggregate/jacoco.xml @@ -93,6 +94,11 @@ kafka-streams ${kafka.version} + + dev.failsafe + failsafe + ${failsafe.version} +