forked from provectus/kafka-ui
-
Notifications
You must be signed in to change notification settings - Fork 1
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
1 parent
162b39d
commit 27156c0
Showing
862 changed files
with
39,326 additions
and
0 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
Oops, something went wrong.
1 change: 1 addition & 0 deletions
1
...18/jacoco/com.provectus.kafka.ui.client/RetryingKafkaConnectClient$RetryingApiClient.html
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 @@ | ||
| <?xml version="1.0" encoding="UTF-8"?><!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd"><html xmlns="http://www.w3.org/1999/xhtml" lang="en"><head><meta http-equiv="Content-Type" content="text/html;charset=UTF-8"/><link rel="stylesheet" href="../jacoco-resources/report.css" type="text/css"/><link rel="shortcut icon" href="../jacoco-resources/report.gif" type="image/gif"/><title>RetryingKafkaConnectClient.RetryingApiClient</title><script type="text/javascript" src="../jacoco-resources/sort.js"></script></head><body onload="initialSort(['breadcrumb'])"><div class="breadcrumb" id="breadcrumb"><span class="info"><a href="../jacoco-sessions.html" class="el_session">Sessions</a></span><a href="../index.html" class="el_report">kafka-ui-api</a> > <a href="index.html" class="el_package">com.provectus.kafka.ui.client</a> > <span class="el_class">RetryingKafkaConnectClient.RetryingApiClient</span></div><h1>RetryingKafkaConnectClient.RetryingApiClient</h1><table class="coverage" cellspacing="0" id="coveragetable"><thead><tr><td class="sortable" id="a" onclick="toggleSort(this)">Element</td><td class="down sortable bar" id="b" onclick="toggleSort(this)">Missed Instructions</td><td class="sortable ctr2" id="c" onclick="toggleSort(this)">Cov.</td><td class="sortable bar" id="d" onclick="toggleSort(this)">Missed Branches</td><td class="sortable ctr2" id="e" onclick="toggleSort(this)">Cov.</td><td class="sortable ctr1" id="f" onclick="toggleSort(this)">Missed</td><td class="sortable ctr2" id="g" onclick="toggleSort(this)">Cxty</td><td class="sortable ctr1" id="h" onclick="toggleSort(this)">Missed</td><td class="sortable ctr2" id="i" onclick="toggleSort(this)">Lines</td><td class="sortable ctr1" id="j" onclick="toggleSort(this)">Missed</td><td class="sortable ctr2" id="k" onclick="toggleSort(this)">Methods</td></tr></thead><tfoot><tr><td>Total</td><td class="bar">0 of 43</td><td class="ctr2">100%</td><td class="bar">0 of 0</td><td class="ctr2">n/a</td><td class="ctr1">0</td><td class="ctr2">2</td><td class="ctr1">0</td><td class="ctr2">14</td><td class="ctr1">0</td><td class="ctr2">2</td></tr></tfoot><tbody><tr><td id="a1"><a href="RetryingKafkaConnectClient.java.html#L259" class="el_method">RetryingKafkaConnectClient.RetryingApiClient(ClustersProperties.ConnectCluster, ClustersProperties.TruststoreConfig, DataSize)</a></td><td class="bar" id="b0"><img src="../jacoco-resources/greenbar.gif" width="120" height="10" title="22" alt="22"/></td><td class="ctr2" id="c0">100%</td><td class="bar" id="d0"/><td class="ctr2" id="e0">n/a</td><td class="ctr1" id="f0">0</td><td class="ctr2" id="g0">1</td><td class="ctr1" id="h0">0</td><td class="ctr2" id="i1">5</td><td class="ctr1" id="j0">0</td><td class="ctr2" id="k0">1</td></tr><tr><td id="a0"><a href="RetryingKafkaConnectClient.java.html#L268" class="el_method">buildWebClient(DataSize, ClustersProperties.ConnectCluster, ClustersProperties.TruststoreConfig)</a></td><td class="bar" id="b1"><img src="../jacoco-resources/greenbar.gif" width="114" height="10" title="21" alt="21"/></td><td class="ctr2" id="c1">100%</td><td class="bar" id="d1"/><td class="ctr2" id="e1">n/a</td><td class="ctr1" id="f1">0</td><td class="ctr2" id="g1">1</td><td class="ctr1" id="h1">0</td><td class="ctr2" id="i0">9</td><td class="ctr1" id="j1">0</td><td class="ctr2" id="k1">1</td></tr></tbody></table><div class="footer"><span class="right">Created with <a href="http://www.jacoco.org/jacoco">JaCoCo</a> 0.8.10.202304240956</span></div></body></html> |
1 change: 1 addition & 0 deletions
1
...orts/report-20240218/jacoco/com.provectus.kafka.ui.client/RetryingKafkaConnectClient.html
Large diffs are not rendered by default.
Oops, something went wrong.
285 changes: 285 additions & 0 deletions
285
...report-20240218/jacoco/com.provectus.kafka.ui.client/RetryingKafkaConnectClient.java.html
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,285 @@ | ||
| <?xml version="1.0" encoding="UTF-8"?><!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd"><html xmlns="http://www.w3.org/1999/xhtml" lang="en"><head><meta http-equiv="Content-Type" content="text/html;charset=UTF-8"/><link rel="stylesheet" href="../jacoco-resources/report.css" type="text/css"/><link rel="shortcut icon" href="../jacoco-resources/report.gif" type="image/gif"/><title>RetryingKafkaConnectClient.java</title><link rel="stylesheet" href="../jacoco-resources/prettify.css" type="text/css"/><script type="text/javascript" src="../jacoco-resources/prettify.js"></script></head><body onload="window['PR_TAB_WIDTH']=4;prettyPrint()"><div class="breadcrumb" id="breadcrumb"><span class="info"><a href="../jacoco-sessions.html" class="el_session">Sessions</a></span><a href="../index.html" class="el_report">kafka-ui-api</a> > <a href="index.source.html" class="el_package">com.provectus.kafka.ui.client</a> > <span class="el_source">RetryingKafkaConnectClient.java</span></div><h1>RetryingKafkaConnectClient.java</h1><pre class="source lang-java linenums">package com.provectus.kafka.ui.client; | ||
|
|
||
| import static com.provectus.kafka.ui.config.ClustersProperties.ConnectCluster; | ||
|
|
||
| import com.provectus.kafka.ui.config.ClustersProperties; | ||
| import com.provectus.kafka.ui.connect.ApiClient; | ||
| import com.provectus.kafka.ui.connect.api.KafkaConnectClientApi; | ||
| import com.provectus.kafka.ui.connect.model.Connector; | ||
| import com.provectus.kafka.ui.connect.model.ConnectorPlugin; | ||
| import com.provectus.kafka.ui.connect.model.ConnectorPluginConfigValidationResponse; | ||
| import com.provectus.kafka.ui.connect.model.ConnectorStatus; | ||
| import com.provectus.kafka.ui.connect.model.ConnectorTask; | ||
| import com.provectus.kafka.ui.connect.model.ConnectorTopics; | ||
| import com.provectus.kafka.ui.connect.model.NewConnector; | ||
| import com.provectus.kafka.ui.connect.model.TaskStatus; | ||
| import com.provectus.kafka.ui.exception.KafkaConnectConflictReponseException; | ||
| import com.provectus.kafka.ui.exception.ValidationException; | ||
| import com.provectus.kafka.ui.util.WebClientConfigurator; | ||
| import java.time.Duration; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import javax.annotation.Nullable; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.http.ResponseEntity; | ||
| import org.springframework.util.unit.DataSize; | ||
| import org.springframework.web.client.RestClientException; | ||
| import org.springframework.web.reactive.function.client.WebClient; | ||
| import org.springframework.web.reactive.function.client.WebClientResponseException; | ||
| import reactor.core.publisher.Flux; | ||
| import reactor.core.publisher.Mono; | ||
| import reactor.util.retry.Retry; | ||
|
|
||
| <span class="fc" id="L33">@Slf4j</span> | ||
| public class RetryingKafkaConnectClient extends KafkaConnectClientApi { | ||
| private static final int MAX_RETRIES = 5; | ||
| <span class="fc" id="L36"> private static final Duration RETRIES_DELAY = Duration.ofMillis(200);</span> | ||
|
|
||
| public RetryingKafkaConnectClient(ConnectCluster config, | ||
| @Nullable ClustersProperties.TruststoreConfig truststoreConfig, | ||
| DataSize maxBuffSize) { | ||
| <span class="fc" id="L41"> super(new RetryingApiClient(config, truststoreConfig, maxBuffSize));</span> | ||
| <span class="fc" id="L42"> }</span> | ||
|
|
||
| private static Retry conflictCodeRetry() { | ||
| <span class="fc" id="L45"> return Retry</span> | ||
| <span class="fc" id="L46"> .fixedDelay(MAX_RETRIES, RETRIES_DELAY)</span> | ||
| <span class="fc" id="L47"> .filter(e -> e instanceof WebClientResponseException.Conflict)</span> | ||
| <span class="fc" id="L48"> .onRetryExhaustedThrow((spec, signal) -></span> | ||
| <span class="nc" id="L49"> new KafkaConnectConflictReponseException(</span> | ||
| <span class="nc" id="L50"> (WebClientResponseException.Conflict) signal.failure()));</span> | ||
| } | ||
|
|
||
| private static <T> Mono<T> withRetryOnConflict(Mono<T> publisher) { | ||
| <span class="fc" id="L54"> return publisher.retryWhen(conflictCodeRetry());</span> | ||
| } | ||
|
|
||
| private static <T> Flux<T> withRetryOnConflict(Flux<T> publisher) { | ||
| <span class="fc" id="L58"> return publisher.retryWhen(conflictCodeRetry());</span> | ||
| } | ||
|
|
||
| private static <T> Mono<T> withBadRequestErrorHandling(Mono<T> publisher) { | ||
| <span class="fc" id="L62"> return publisher</span> | ||
| <span class="fc" id="L63"> .onErrorResume(WebClientResponseException.BadRequest.class, e -></span> | ||
| <span class="fc" id="L64"> Mono.error(new ValidationException("Invalid configuration")))</span> | ||
| <span class="fc" id="L65"> .onErrorResume(WebClientResponseException.InternalServerError.class, e -></span> | ||
| <span class="fc" id="L66"> Mono.error(new ValidationException("Invalid configuration")));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Connector> createConnector(NewConnector newConnector) throws RestClientException { | ||
| <span class="fc" id="L71"> return withBadRequestErrorHandling(</span> | ||
| <span class="fc" id="L72"> super.createConnector(newConnector)</span> | ||
| ); | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Connector> setConnectorConfig(String connectorName, Map<String, Object> requestBody) | ||
| throws RestClientException { | ||
| <span class="fc" id="L79"> return withBadRequestErrorHandling(</span> | ||
| <span class="fc" id="L80"> super.setConnectorConfig(connectorName, requestBody)</span> | ||
| ); | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Connector>> createConnectorWithHttpInfo(NewConnector newConnector) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L87"> return withRetryOnConflict(super.createConnectorWithHttpInfo(newConnector));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Void> deleteConnector(String connectorName) throws WebClientResponseException { | ||
| <span class="fc" id="L92"> return withRetryOnConflict(super.deleteConnector(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Void>> deleteConnectorWithHttpInfo(String connectorName) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L98"> return withRetryOnConflict(super.deleteConnectorWithHttpInfo(connectorName));</span> | ||
| } | ||
|
|
||
|
|
||
| @Override | ||
| public Mono<Connector> getConnector(String connectorName) throws WebClientResponseException { | ||
| <span class="fc" id="L104"> return withRetryOnConflict(super.getConnector(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Connector>> getConnectorWithHttpInfo(String connectorName) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L110"> return withRetryOnConflict(super.getConnectorWithHttpInfo(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Map<String, Object>> getConnectorConfig(String connectorName) throws WebClientResponseException { | ||
| <span class="fc" id="L115"> return withRetryOnConflict(super.getConnectorConfig(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Map<String, Object>>> getConnectorConfigWithHttpInfo(String connectorName) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L121"> return withRetryOnConflict(super.getConnectorConfigWithHttpInfo(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Flux<ConnectorPlugin> getConnectorPlugins() throws WebClientResponseException { | ||
| <span class="fc" id="L126"> return withRetryOnConflict(super.getConnectorPlugins());</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<List<ConnectorPlugin>>> getConnectorPluginsWithHttpInfo() | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L132"> return withRetryOnConflict(super.getConnectorPluginsWithHttpInfo());</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ConnectorStatus> getConnectorStatus(String connectorName) throws WebClientResponseException { | ||
| <span class="fc" id="L137"> return withRetryOnConflict(super.getConnectorStatus(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<ConnectorStatus>> getConnectorStatusWithHttpInfo(String connectorName) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L143"> return withRetryOnConflict(super.getConnectorStatusWithHttpInfo(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<TaskStatus> getConnectorTaskStatus(String connectorName, Integer taskId) | ||
| throws WebClientResponseException { | ||
| <span class="fc" id="L149"> return withRetryOnConflict(super.getConnectorTaskStatus(connectorName, taskId));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<TaskStatus>> getConnectorTaskStatusWithHttpInfo(String connectorName, Integer taskId) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L155"> return withRetryOnConflict(super.getConnectorTaskStatusWithHttpInfo(connectorName, taskId));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Flux<ConnectorTask> getConnectorTasks(String connectorName) throws WebClientResponseException { | ||
| <span class="fc" id="L160"> return withRetryOnConflict(super.getConnectorTasks(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<List<ConnectorTask>>> getConnectorTasksWithHttpInfo(String connectorName) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L166"> return withRetryOnConflict(super.getConnectorTasksWithHttpInfo(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Map<String, ConnectorTopics>> getConnectorTopics(String connectorName) throws WebClientResponseException { | ||
| <span class="fc" id="L171"> return withRetryOnConflict(super.getConnectorTopics(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Map<String, ConnectorTopics>>> getConnectorTopicsWithHttpInfo(String connectorName) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L177"> return withRetryOnConflict(super.getConnectorTopicsWithHttpInfo(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Flux<String> getConnectors(String search) throws WebClientResponseException { | ||
| <span class="fc" id="L182"> return withRetryOnConflict(super.getConnectors(search));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<List<String>>> getConnectorsWithHttpInfo(String search) throws WebClientResponseException { | ||
| <span class="nc" id="L187"> return withRetryOnConflict(super.getConnectorsWithHttpInfo(search));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Void> pauseConnector(String connectorName) throws WebClientResponseException { | ||
| <span class="nc" id="L192"> return withRetryOnConflict(super.pauseConnector(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Void>> pauseConnectorWithHttpInfo(String connectorName) throws WebClientResponseException { | ||
| <span class="nc" id="L197"> return withRetryOnConflict(super.pauseConnectorWithHttpInfo(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Void> restartConnector(String connectorName, Boolean includeTasks, Boolean onlyFailed) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L203"> return withRetryOnConflict(super.restartConnector(connectorName, includeTasks, onlyFailed));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Void>> restartConnectorWithHttpInfo(String connectorName, Boolean includeTasks, | ||
| Boolean onlyFailed) throws WebClientResponseException { | ||
| <span class="nc" id="L209"> return withRetryOnConflict(super.restartConnectorWithHttpInfo(connectorName, includeTasks, onlyFailed));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Void> restartConnectorTask(String connectorName, Integer taskId) throws WebClientResponseException { | ||
| <span class="nc" id="L214"> return withRetryOnConflict(super.restartConnectorTask(connectorName, taskId));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Void>> restartConnectorTaskWithHttpInfo(String connectorName, Integer taskId) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L220"> return withRetryOnConflict(super.restartConnectorTaskWithHttpInfo(connectorName, taskId));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Void> resumeConnector(String connectorName) throws WebClientResponseException { | ||
| <span class="nc" id="L225"> return super.resumeConnector(connectorName);</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Void>> resumeConnectorWithHttpInfo(String connectorName) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L231"> return withRetryOnConflict(super.resumeConnectorWithHttpInfo(connectorName));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<Connector>> setConnectorConfigWithHttpInfo(String connectorName, | ||
| Map<String, Object> requestBody) | ||
| throws WebClientResponseException { | ||
| <span class="nc" id="L238"> return withRetryOnConflict(super.setConnectorConfigWithHttpInfo(connectorName, requestBody));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ConnectorPluginConfigValidationResponse> validateConnectorPluginConfig(String pluginName, | ||
| Map<String, Object> requestBody) | ||
| throws WebClientResponseException { | ||
| <span class="fc" id="L245"> return withRetryOnConflict(super.validateConnectorPluginConfig(pluginName, requestBody));</span> | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<ResponseEntity<ConnectorPluginConfigValidationResponse>> validateConnectorPluginConfigWithHttpInfo( | ||
| String pluginName, Map<String, Object> requestBody) throws WebClientResponseException { | ||
| <span class="nc" id="L251"> return withRetryOnConflict(super.validateConnectorPluginConfigWithHttpInfo(pluginName, requestBody));</span> | ||
| } | ||
|
|
||
| private static class RetryingApiClient extends ApiClient { | ||
|
|
||
| public RetryingApiClient(ConnectCluster config, | ||
| ClustersProperties.TruststoreConfig truststoreConfig, | ||
| DataSize maxBuffSize) { | ||
| <span class="fc" id="L259"> super(buildWebClient(maxBuffSize, config, truststoreConfig), null, null);</span> | ||
| <span class="fc" id="L260"> setBasePath(config.getAddress());</span> | ||
| <span class="fc" id="L261"> setUsername(config.getUsername());</span> | ||
| <span class="fc" id="L262"> setPassword(config.getPassword());</span> | ||
| <span class="fc" id="L263"> }</span> | ||
|
|
||
| public static WebClient buildWebClient(DataSize maxBuffSize, | ||
| ConnectCluster config, | ||
| ClustersProperties.TruststoreConfig truststoreConfig) { | ||
| <span class="fc" id="L268"> return new WebClientConfigurator()</span> | ||
| <span class="fc" id="L269"> .configureSsl(</span> | ||
| truststoreConfig, | ||
| new ClustersProperties.KeystoreConfig( | ||
| <span class="fc" id="L272"> config.getKeystoreLocation(),</span> | ||
| <span class="fc" id="L273"> config.getKeystorePassword()</span> | ||
| ) | ||
| ) | ||
| <span class="fc" id="L276"> .configureBasicAuth(</span> | ||
| <span class="fc" id="L277"> config.getUsername(),</span> | ||
| <span class="fc" id="L278"> config.getPassword()</span> | ||
| ) | ||
| <span class="fc" id="L280"> .configureBufferSize(maxBuffSize)</span> | ||
| <span class="fc" id="L281"> .build();</span> | ||
| } | ||
| } | ||
| } | ||
| </pre><div class="footer"><span class="right">Created with <a href="http://www.jacoco.org/jacoco">JaCoCo</a> 0.8.10.202304240956</span></div></body></html> |
Oops, something went wrong.