Merge branch 'master' of github.com:provectus/kafka-ui into issues/2728

Author: Oleg Sergeev

.github/workflows/e2e-checks.yaml

@@ -76,4 +76,4 @@ jobs:
           target_url: https://${{ github.repository_owner }}.github.io/kafka-ui/allure/allure-results/${{ github.run_number }}
       - name: Dump docker logs on failure
         if: failure()
-        uses: jwalton/[email protected].0
+        uses: jwalton/[email protected].1

.github/workflows/frontend.yaml

@@ -20,11 +20,11 @@ jobs:
           # Disabling shallow clone is recommended for improving relevancy of reporting
           fetch-depth: 0
           ref: ${{ github.event.pull_request.head.sha }}
-      - uses: pnpm/[email protected].2
+      - uses: pnpm/[email protected].4
         with:
           version: 7.4.0
       - name: Install node
-        uses: actions/setup-node@v3.4.1
+        uses: actions/setup-node@v3.5.1
         with:
           node-version: "16.15.0"
           cache: "pnpm"

.github/workflows/helm.yaml

@@ -1,23 +1,29 @@
 name: Helm
 on:
-  pull_request:
-    types: [ 'opened', 'edited', 'reopened', 'synchronize' ]
-    paths:
-      - "charts/**"
-
-  schedule:
-    # * is a special character in YAML so you have to quote this string
-    - cron:  '0 8 * * 3'
-
+ pull_request:
+  types: ["opened", "edited", "reopened", "synchronize"]
+  branches:
+   - 'master'
+  paths:
+   - "charts/**"
 jobs:
   build-and-test:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
       - name: Helm tool installer
-        uses: Azure/setup-helm@v1
+        uses: Azure/setup-helm@v3
       - name: Setup Kubeval
         uses: lra/[email protected]
+      #check, was helm version increased in Chart.yaml?
+      - name: Check version
+        shell: bash
+        run: |
+          helm_version_new=$(cat charts/kafka-ui/Chart.yaml  | grep version | awk  '{print $2}')
+          helm_version_old=$(curl -s https://raw.githubusercontent.com/provectus/kafka-ui/master/charts/kafka-ui/Chart.yaml |   grep version | awk  '{print $2}' )
+          echo $helm_version_old
+          echo $helm_version_new
+          if [[ "$helm_version_new" > "$helm_version_old" ]]; then exit 0 ; else exit 1 ; fi
       - name: Run kubeval
         shell: bash
         run: |

.github/workflows/release-helm.yaml

@@ -0,0 +1,37 @@
+name: Release helm
+on:
+ push:
+    branches:
+     - master
+    paths:
+      - "charts/**"
+
+jobs:
+ release-helm:
+  runs-on:
+   ubuntu-latest
+  steps:
+      - uses: actions/checkout@v3
+        with:
+          fetch-depth: 1
+
+      - run: |
+          git config user.name github-actions
+          git config user.email [email protected]
+
+      - uses: azure/setup-helm@v1
+
+      - name: add chart #realse helm with new version
+        run: |
+          echo "VERSION=$(cat charts/kafka-ui/Chart.yaml  | grep version | awk '{print $2}')" >> $GITHUB_ENV
+          MSG=$(helm package  charts/kafka-ui)
+          git fetch origin
+          git stash
+          git checkout -b gh-pages origin/gh-pages
+          helm repo index .
+          git add -f ${MSG##*/} index.yaml
+          git commit -m "release ${{ env.VERSION }}"
+          git push
+      - uses: rickstaa/action-create-tag@v1 #create new tag
+        with:
+          tag: "charts/kafka-ui-${{ env.VERSION }}"

.github/workflows/stale.yaml

@@ -7,7 +7,7 @@ jobs:
   stale:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/stale@v5
+      - uses: actions/stale@v6
         with:
           days-before-issue-stale: 7
           days-before-issue-close: 3

charts/kafka-ui/Chart.yaml

@@ -2,6 +2,6 @@ apiVersion: v2
 name: kafka-ui
 description: A Helm chart for kafka-UI
 type: application
-version: 0.4.3
-appVersion: latest
+version: 0.4.4
+appVersion: 0.4.0
 icon: https://github.com/provectus/kafka-ui/raw/master/documentation/images/kafka-ui-logo.png

kafka-ui-api/pom.xml

@@ -173,6 +173,12 @@
             <version>${mockito.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>net.bytebuddy</groupId>
+            <artifactId>byte-buddy</artifactId>
+            <version>${byte-buddy.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>

kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KsqlClient.java

@@ -3,13 +3,16 @@
 import com.provectus.kafka.ui.api.KafkaConnectApi;
 import com.provectus.kafka.ui.model.ConnectDTO;
 import com.provectus.kafka.ui.model.ConnectorActionDTO;
+import com.provectus.kafka.ui.model.ConnectorColumnsToSortDTO;
 import com.provectus.kafka.ui.model.ConnectorDTO;
 import com.provectus.kafka.ui.model.ConnectorPluginConfigValidationResponseDTO;
 import com.provectus.kafka.ui.model.ConnectorPluginDTO;
 import com.provectus.kafka.ui.model.FullConnectorInfoDTO;
 import com.provectus.kafka.ui.model.NewConnectorDTO;
+import com.provectus.kafka.ui.model.SortOrderDTO;
 import com.provectus.kafka.ui.model.TaskDTO;
 import com.provectus.kafka.ui.service.KafkaConnectService;
+import java.util.Comparator;
 import java.util.Map;
 import javax.validation.Valid;
 import lombok.RequiredArgsConstructor;
@@ -68,10 +71,16 @@ public Mono<ResponseEntity<Void>> deleteConnector(String clusterName, String con
   public Mono<ResponseEntity<Flux<FullConnectorInfoDTO>>> getAllConnectors(
       String clusterName,
       String search,
+      ConnectorColumnsToSortDTO orderBy,
+      SortOrderDTO sortOrder,
       ServerWebExchange exchange
   ) {
+    var comparator = sortOrder == null || sortOrder.equals(SortOrderDTO.ASC)
+            ? getConnectorsComparator(orderBy)
+            : getConnectorsComparator(orderBy).reversed();
     return Mono.just(ResponseEntity.ok(
-        kafkaConnectService.getAllConnectors(getCluster(clusterName), search)));
+        kafkaConnectService.getAllConnectors(getCluster(clusterName), search).sort(comparator))
+    );
   }
 
   @Override
@@ -142,4 +151,22 @@ public Mono<ResponseEntity<Flux<ConnectorPluginDTO>>> getConnectorPlugins(
             getCluster(clusterName), connectName, pluginName, requestBody)
         .map(ResponseEntity::ok);
   }
+
+  private Comparator<FullConnectorInfoDTO> getConnectorsComparator(ConnectorColumnsToSortDTO orderBy) {
+    var defaultComparator = Comparator.comparing(FullConnectorInfoDTO::getName);
+    if (orderBy == null) {
+      return defaultComparator;
+    }
+    switch (orderBy) {
+      case CONNECT:
+        return Comparator.comparing(FullConnectorInfoDTO::getConnect);
+      case TYPE:
+        return Comparator.comparing(FullConnectorInfoDTO::getType);
+      case STATUS:
+        return Comparator.comparing(fullConnectorInfoDTO -> fullConnectorInfoDTO.getStatus().getState());
+      case NAME:
+      default:
+        return defaultComparator;
+    }
+  }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/KafkaConnectController.java

@@ -1,15 +1,12 @@
 package com.provectus.kafka.ui.controller;
 
 import com.provectus.kafka.ui.api.KsqlApi;
-import com.provectus.kafka.ui.model.KsqlCommandDTO;
-import com.provectus.kafka.ui.model.KsqlCommandResponseDTO;
 import com.provectus.kafka.ui.model.KsqlCommandV2DTO;
 import com.provectus.kafka.ui.model.KsqlCommandV2ResponseDTO;
 import com.provectus.kafka.ui.model.KsqlResponseDTO;
 import com.provectus.kafka.ui.model.KsqlStreamDescriptionDTO;
 import com.provectus.kafka.ui.model.KsqlTableDescriptionDTO;
 import com.provectus.kafka.ui.model.KsqlTableResponseDTO;
-import com.provectus.kafka.ui.service.KsqlService;
 import com.provectus.kafka.ui.service.ksql.KsqlServiceV2;
 import java.util.List;
 import java.util.Map;
@@ -27,17 +24,8 @@
 @RequiredArgsConstructor
 @Slf4j
 public class KsqlController extends AbstractController implements KsqlApi {
-  private final KsqlService ksqlService;
-  private final KsqlServiceV2 ksqlServiceV2;
 
-  @Override
-  public Mono<ResponseEntity<KsqlCommandResponseDTO>> executeKsqlCommand(String clusterName,
-                                                                         Mono<KsqlCommandDTO>
-                                                                             ksqlCommand,
-                                                                         ServerWebExchange exchange) {
-    return ksqlService.executeKsqlCommand(getCluster(clusterName), ksqlCommand)
-        .map(ResponseEntity::ok);
-  }
+  private final KsqlServiceV2 ksqlServiceV2;
 
   @Override
   public Mono<ResponseEntity<KsqlCommandV2ResponseDTO>> executeKsql(String clusterName,

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/KsqlController.java

@@ -5,6 +5,7 @@
 import static java.util.stream.Collectors.toMap;
 
 import com.provectus.kafka.ui.api.MessagesApi;
+import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.model.ConsumerPosition;
 import com.provectus.kafka.ui.model.CreateTopicMessageDTO;
 import com.provectus.kafka.ui.model.MessageFilterTypeDTO;
@@ -18,6 +19,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import javax.annotation.Nullable;
 import javax.validation.Valid;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -63,18 +65,22 @@ public Mono<ResponseEntity<Flux<TopicMessageEventDTO>>> getTopicMessages(String
                                                                            String keySerde,
                                                                            String valueSerde,
                                                                            ServerWebExchange exchange) {
+    seekType = seekType != null ? seekType : SeekTypeDTO.BEGINNING;
+    seekDirection = seekDirection != null ? seekDirection : SeekDirectionDTO.FORWARD;
+    filterQueryType = filterQueryType != null ? filterQueryType : MessageFilterTypeDTO.STRING_CONTAINS;
+    int recordsLimit =
+        Optional.ofNullable(limit).map(s -> Math.min(s, MAX_LOAD_RECORD_LIMIT)).orElse(DEFAULT_LOAD_RECORD_LIMIT);
+
     var positions = new ConsumerPosition(
-        seekType != null ? seekType : SeekTypeDTO.BEGINNING,
-        parseSeekTo(topicName, seekTo),
-        seekDirection
+        seekType,
+        topicName,
+        parseSeekTo(topicName, seekType, seekTo)
     );
-    int recordsLimit = Optional.ofNullable(limit)
-        .map(s -> Math.min(s, MAX_LOAD_RECORD_LIMIT))
-        .orElse(DEFAULT_LOAD_RECORD_LIMIT);
     return Mono.just(
         ResponseEntity.ok(
             messagesService.loadMessages(
-                getCluster(clusterName), topicName, positions, q, filterQueryType, recordsLimit, keySerde, valueSerde)
+                getCluster(clusterName), topicName, positions, q, filterQueryType,
+                recordsLimit, seekDirection, keySerde, valueSerde)
         )
     );
   }
@@ -92,9 +98,13 @@ public Mono<ResponseEntity<Void>> sendTopicMessages(
    * The format is [partition]::[offset] for specifying offsets
    * or [partition]::[timestamp in millis] for specifying timestamps.
    */
-  private Map<TopicPartition, Long> parseSeekTo(String topic, List<String> seekTo) {
+  @Nullable
+  private Map<TopicPartition, Long> parseSeekTo(String topic, SeekTypeDTO seekType, List<String> seekTo) {
     if (seekTo == null || seekTo.isEmpty()) {
-      return Map.of();
+      if (seekType == SeekTypeDTO.LATEST || seekType == SeekTypeDTO.BEGINNING) {
+        return null;
+      }
+      throw new ValidationException("seekTo should be set if seekType is " + seekType);
     }
     return seekTo.stream()
         .map(p -> {