BE: Chore: Upgrade kafka to confluent 7.8.0

.dev/dev_arm64.yaml

@@ -32,7 +32,7 @@ services:
  KAFKA_CLUSTERS_0_AUDIT_CONSOLEAUDITENABLED: 'true'
 
  kafka0:
- image: confluentinc/cp-kafka:7.6.0.arm64
+ image: confluentinc/cp-kafka:7.8.0.arm64
  user: "0:0"
  hostname: kafka0
  container_name: kafka0
@@ -60,7 +60,7 @@ services:
  CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
 
  schema-registry0:
- image: confluentinc/cp-schema-registry:7.6.0.arm64
+ image: confluentinc/cp-schema-registry:7.8.0.arm64
  ports:
  - 8085:8085
  depends_on:
@@ -76,7 +76,7 @@ services:
  SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
 
  kafka-connect0:
- image: confluentinc/cp-kafka-connect:7.6.0.arm64
+ image: confluentinc/cp-kafka-connect:7.8.0.arm64
  ports:
  - 8083:8083
  depends_on:
@@ -101,7 +101,7 @@ services:
  CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components,/usr/local/share/kafka/plugins,/usr/share/filestream-connectors"
 
  ksqldb0:
- image: confluentinc/cp-ksqldb-server:7.6.0.arm64
+ image: confluentinc/cp-ksqldb-server:7.8.0.arm64
  depends_on:
  - kafka0
  - kafka-connect0
@@ -119,7 +119,7 @@ services:
  KSQL_CACHE_MAX_BYTES_BUFFERING: 0
 
  kafka-init-topics:
- image: confluentinc/cp-kafka:7.6.0.arm64
+ image: confluentinc/cp-kafka:7.8.0.arm64
  volumes:
  - ../documentation/compose/data/message.json:/data/message.json
  depends_on:

.gitignore

@@ -42,3 +42,4 @@ build/
 *.tgz
 
 /docker/*.override.yaml
+/e2e-tests/allure-results/

api/pom.xml

@@ -50,7 +50,10 @@
  <dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
- <version>${kafka-clients.version}</version>
+ <!-- ccs stands for Confluent Community Edition
+ See https://www.confluent.io/confluent-community-license-faq/
+ -->
+ <version>${confluent.version}-ccs</version>
  </dependency>
  <dependency>
  <groupId>org.apache.commons</groupId>

api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java

@@ -22,6 +22,7 @@
 import java.util.Objects;
 import javax.annotation.Nullable;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.http.ResponseEntity;
 import org.springframework.util.unit.DataSize;
 import org.springframework.web.client.RestClientException;
@@ -51,14 +52,36 @@ private static Retry conflictCodeRetry() {
  (WebClientResponseException.Conflict) signal.failure()));
  }
 
- private static <T> Mono<T> withRetryOnConflict(Mono<T> publisher) {
- return publisher.retryWhen(conflictCodeRetry());
+ private static @NotNull Retry retryOnRebalance() {
+ return Retry.fixedDelay(MAX_RETRIES, RETRIES_DELAY).filter(e -> {
+
+ if (e instanceof WebClientResponseException.InternalServerError exception) {
+ final var errorMessage = getMessage(exception);
+ return StringUtils.equals(errorMessage,
+ // From https://github.com/apache/kafka/blob/dfc07e0e0c6e737a56a5402644265f634402b864/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L2340
+ "Request cannot be completed because a rebalance is expected");
+ }
+ return false;
+ });
+ }
+
+ private static <T> Mono<T> withRetryOnConflictOrRebalance(Mono<T> publisher) {
+ return publisher
+ .retryWhen(retryOnRebalance())
+ .retryWhen(conflictCodeRetry());
+ }
+
+ private static <T> Flux<T> withRetryOnConflictOrRebalance(Flux<T> publisher) {
+ return publisher
+ .retryWhen(retryOnRebalance())
+ .retryWhen(conflictCodeRetry());
  }
 
- private static <T> Flux<T> withRetryOnConflict(Flux<T> publisher) {
- return publisher.retryWhen(conflictCodeRetry());
+ private static <T> Mono<T> withRetryOnRebalance(Mono<T> publisher) {
+ return publisher.retryWhen(retryOnRebalance());
  }
 
+
  private static <T> Mono<T> withBadRequestErrorHandling(Mono<T> publisher) {
  return publisher
  .onErrorResume(WebClientResponseException.BadRequest.class,
@@ -73,197 +96,200 @@ private record ErrorMessage(@NotNull @JsonProperty("message") String message) {
  }
 
  private static <T> @NotNull Mono<T> parseConnectErrorMessage(WebClientResponseException parseException) {
+ return Mono.error(new ValidationException(getMessage(parseException)));
+ }
+
+ private static String getMessage(WebClientResponseException parseException) {
  final var errorMessage = parseException.getResponseBodyAs(ErrorMessage.class);
- return Mono.error(new ValidationException(
- Objects.requireNonNull(errorMessage,
- // see https://github.com/apache/kafka/blob/a0a501952b6d61f6f273bdb8f842346b51e9dfce/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/errors/ConnectExceptionMapper.java
- "This should not happen according to the ConnectExceptionMapper")
- .message()));
+ return Objects.requireNonNull(errorMessage,
+ // see https://github.com/apache/kafka/blob/a0a501952b6d61f6f273bdb8f842346b51e9dfce/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/errors/ConnectExceptionMapper.java
+ "This should not happen according to the ConnectExceptionMapper")
+ .message();
  }
 
  @Override
  public Mono<Connector> createConnector(NewConnector newConnector) throws RestClientException {
  return withBadRequestErrorHandling(
- super.createConnector(newConnector)
+ withRetryOnRebalance(super.createConnector(newConnector))
  );
  }
 
  @Override
  public Mono<Connector> setConnectorConfig(String connectorName, Map<String, Object> requestBody)
  throws RestClientException {
  return withBadRequestErrorHandling(
- super.setConnectorConfig(connectorName, requestBody)
+ withRetryOnRebalance(super.setConnectorConfig(connectorName, requestBody))
  );
  }
 
  @Override
  public Mono<ResponseEntity<Connector>> createConnectorWithHttpInfo(NewConnector newConnector)
  throws WebClientResponseException {
- return withRetryOnConflict(super.createConnectorWithHttpInfo(newConnector));
+ return withRetryOnConflictOrRebalance(super.createConnectorWithHttpInfo(newConnector));
  }
 
  @Override
  public Mono<Void> deleteConnector(String connectorName) throws WebClientResponseException {
- return withRetryOnConflict(super.deleteConnector(connectorName));
+ return withRetryOnConflictOrRebalance(super.deleteConnector(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<Void>> deleteConnectorWithHttpInfo(String connectorName)
  throws WebClientResponseException {
- return withRetryOnConflict(super.deleteConnectorWithHttpInfo(connectorName));
+ return withRetryOnConflictOrRebalance(super.deleteConnectorWithHttpInfo(connectorName));
  }
 
 
  @Override
  public Mono<Connector> getConnector(String connectorName) throws WebClientResponseException {
- return withRetryOnConflict(super.getConnector(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnector(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<Connector>> getConnectorWithHttpInfo(String connectorName)
  throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorWithHttpInfo(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorWithHttpInfo(connectorName));
  }
 
  @Override
  public Mono<Map<String, Object>> getConnectorConfig(String connectorName) throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorConfig(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorConfig(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<Map<String, Object>>> getConnectorConfigWithHttpInfo(String connectorName)
  throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorConfigWithHttpInfo(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorConfigWithHttpInfo(connectorName));
  }
 
  @Override
  public Flux<ConnectorPlugin> getConnectorPlugins() throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorPlugins());
+ return withRetryOnConflictOrRebalance(super.getConnectorPlugins());
  }
 
  @Override
  public Mono<ResponseEntity<List<ConnectorPlugin>>> getConnectorPluginsWithHttpInfo()
  throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorPluginsWithHttpInfo());
+ return withRetryOnConflictOrRebalance(super.getConnectorPluginsWithHttpInfo());
  }
 
  @Override
  public Mono<ConnectorStatus> getConnectorStatus(String connectorName) throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorStatus(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorStatus(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<ConnectorStatus>> getConnectorStatusWithHttpInfo(String connectorName)
  throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorStatusWithHttpInfo(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorStatusWithHttpInfo(connectorName));
  }
 
  @Override
  public Mono<TaskStatus> getConnectorTaskStatus(String connectorName, Integer taskId)
  throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorTaskStatus(connectorName, taskId));
+ return withRetryOnConflictOrRebalance(super.getConnectorTaskStatus(connectorName, taskId));
  }
 
  @Override
  public Mono<ResponseEntity<TaskStatus>> getConnectorTaskStatusWithHttpInfo(String connectorName, Integer taskId)
  throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorTaskStatusWithHttpInfo(connectorName, taskId));
+ return withRetryOnConflictOrRebalance(super.getConnectorTaskStatusWithHttpInfo(connectorName, taskId));
  }
 
  @Override
  public Flux<ConnectorTask> getConnectorTasks(String connectorName) throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorTasks(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorTasks(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<List<ConnectorTask>>> getConnectorTasksWithHttpInfo(String connectorName)
  throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorTasksWithHttpInfo(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorTasksWithHttpInfo(connectorName));
  }
 
  @Override
  public Mono<Map<String, ConnectorTopics>> getConnectorTopics(String connectorName) throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorTopics(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorTopics(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<Map<String, ConnectorTopics>>> getConnectorTopicsWithHttpInfo(String connectorName)
  throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorTopicsWithHttpInfo(connectorName));
+ return withRetryOnConflictOrRebalance(super.getConnectorTopicsWithHttpInfo(connectorName));
  }
 
  @Override
  public Mono<List<String>> getConnectors(String search) throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectors(search));
+ return withRetryOnConflictOrRebalance(super.getConnectors(search));
  }
 
  @Override
  public Mono<ResponseEntity<List<String>>> getConnectorsWithHttpInfo(String search) throws WebClientResponseException {
- return withRetryOnConflict(super.getConnectorsWithHttpInfo(search));
+ return withRetryOnConflictOrRebalance(super.getConnectorsWithHttpInfo(search));
  }
 
  @Override
  public Mono<Void> pauseConnector(String connectorName) throws WebClientResponseException {
- return withRetryOnConflict(super.pauseConnector(connectorName));
+ return withRetryOnConflictOrRebalance(super.pauseConnector(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<Void>> pauseConnectorWithHttpInfo(String connectorName) throws WebClientResponseException {
- return withRetryOnConflict(super.pauseConnectorWithHttpInfo(connectorName));
+ return withRetryOnConflictOrRebalance(super.pauseConnectorWithHttpInfo(connectorName));
  }
 
  @Override
  public Mono<Void> restartConnector(String connectorName, Boolean includeTasks, Boolean onlyFailed)
  throws WebClientResponseException {
- return withRetryOnConflict(super.restartConnector(connectorName, includeTasks, onlyFailed));
+ return withRetryOnConflictOrRebalance(super.restartConnector(connectorName, includeTasks, onlyFailed));
  }
 
  @Override
  public Mono<ResponseEntity<Void>> restartConnectorWithHttpInfo(String connectorName, Boolean includeTasks,
  Boolean onlyFailed) throws WebClientResponseException {
- return withRetryOnConflict(super.restartConnectorWithHttpInfo(connectorName, includeTasks, onlyFailed));
+ return withRetryOnConflictOrRebalance(super.restartConnectorWithHttpInfo(connectorName, includeTasks, onlyFailed));
  }
 
  @Override
  public Mono<Void> restartConnectorTask(String connectorName, Integer taskId) throws WebClientResponseException {
- return withRetryOnConflict(super.restartConnectorTask(connectorName, taskId));
+ return withRetryOnConflictOrRebalance(super.restartConnectorTask(connectorName, taskId));
  }
 
  @Override
  public Mono<ResponseEntity<Void>> restartConnectorTaskWithHttpInfo(String connectorName, Integer taskId)
  throws WebClientResponseException {
- return withRetryOnConflict(super.restartConnectorTaskWithHttpInfo(connectorName, taskId));
+ return withRetryOnConflictOrRebalance(super.restartConnectorTaskWithHttpInfo(connectorName, taskId));
  }
 
  @Override
  public Mono<Void> resumeConnector(String connectorName) throws WebClientResponseException {
- return super.resumeConnector(connectorName);
+ return withRetryOnRebalance(super.resumeConnector(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<Void>> resumeConnectorWithHttpInfo(String connectorName)
  throws WebClientResponseException {
- return withRetryOnConflict(super.resumeConnectorWithHttpInfo(connectorName));
+ return withRetryOnConflictOrRebalance(super.resumeConnectorWithHttpInfo(connectorName));
  }
 
  @Override
  public Mono<ResponseEntity<Connector>> setConnectorConfigWithHttpInfo(String connectorName,
  Map<String, Object> requestBody)
  throws WebClientResponseException {
- return withRetryOnConflict(super.setConnectorConfigWithHttpInfo(connectorName, requestBody));
+ return withRetryOnConflictOrRebalance(super.setConnectorConfigWithHttpInfo(connectorName, requestBody));
  }
 
  @Override
  public Mono<ConnectorPluginConfigValidationResponse> validateConnectorPluginConfig(String pluginName,
  Map<String, Object> requestBody)
  throws WebClientResponseException {
- return withRetryOnConflict(super.validateConnectorPluginConfig(pluginName, requestBody));
+ return withRetryOnConflictOrRebalance(super.validateConnectorPluginConfig(pluginName, requestBody));
  }
 
  @Override
  public Mono<ResponseEntity<ConnectorPluginConfigValidationResponse>> validateConnectorPluginConfigWithHttpInfo(
  String pluginName, Map<String, Object> requestBody) throws WebClientResponseException {
- return withRetryOnConflict(super.validateConnectorPluginConfigWithHttpInfo(pluginName, requestBody));
+ return withRetryOnConflictOrRebalance(super.validateConnectorPluginConfigWithHttpInfo(pluginName, requestBody));
  }
 
  private static class RetryingApiClient extends ApiClient {

api/src/test/java/io/kafbat/ui/AbstractIntegrationTest.java

@@ -40,7 +40,7 @@ public abstract class AbstractIntegrationTest {
  private static final boolean IS_ARM =
  System.getProperty("os.arch").contains("arm") || System.getProperty("os.arch").contains("aarch64");
 
- private static final String CONFLUENT_PLATFORM_VERSION = IS_ARM ? "7.2.1.arm64" : "7.2.1";
+ private static final String CONFLUENT_PLATFORM_VERSION = IS_ARM ? "7.8.0.arm64" : "7.8.0";
 
  public static final KafkaContainer kafka = new KafkaContainer(
  DockerImageName.parse("confluentinc/cp-kafka").withTag(CONFLUENT_PLATFORM_VERSION))