Merge branch 'main' into kafbat/203

kafbat · Aug 7, 2024 · 6d31b98 · 6d31b98
2 parents 7843560 + 654978b
commit 6d31b98
Show file tree

Hide file tree

Showing 12 changed files with 101 additions and 114 deletions.
diff --git a/api/src/main/java/io/kafbat/ui/emitter/MessageFilters.java b/api/src/main/java/io/kafbat/ui/emitter/MessageFilters.java
@@ -18,12 +18,12 @@
 import dev.cel.common.types.StructType;
 import dev.cel.compiler.CelCompiler;
 import dev.cel.compiler.CelCompilerFactory;
+import dev.cel.extensions.CelExtensions;
 import dev.cel.parser.CelStandardMacro;
 import dev.cel.runtime.CelEvaluationException;
 import dev.cel.runtime.CelRuntime;
 import dev.cel.runtime.CelRuntimeFactory;
 import io.kafbat.ui.exception.CelException;
-import io.kafbat.ui.model.MessageFilterTypeDTO;
 import io.kafbat.ui.model.TopicMessageDTO;
 import java.util.HashMap;
 import java.util.Map;
@@ -42,8 +42,7 @@ public class MessageFilters {
   private static final String CEL_RECORD_TYPE_NAME = TopicMessageDTO.class.getSimpleName();
 
   private static final CelCompiler CEL_COMPILER = createCompiler();
-  private static final CelRuntime CEL_RUNTIME = CelRuntimeFactory.standardCelRuntimeBuilder()
-      .build();
+  private static final CelRuntime CEL_RUNTIME = createRuntime();
 
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
@@ -143,6 +142,7 @@ private static CelCompiler createCompiler() {
     return CelCompilerFactory.standardCelCompilerBuilder()
         .setOptions(CelOptions.DEFAULT)
         .setStandardMacros(CelStandardMacro.STANDARD_MACROS)
+        .addLibraries(CelExtensions.strings(), CelExtensions.encoders())
         .addVar(CEL_RECORD_VAR_NAME, recordType)
         .setResultType(SimpleType.BOOL)
         .setTypeProvider(new CelTypeProvider() {
@@ -159,6 +159,12 @@ public Optional<CelType> findType(String typeName) {
         .build();
   }
 
+  private static CelRuntime createRuntime() {
+    return CelRuntimeFactory.standardCelRuntimeBuilder()
+        .addLibraries(CelExtensions.strings(), CelExtensions.encoders())
+        .build();
+  }
+
   @Nullable
   private static Object parseToJsonOrReturnAsIs(@Nullable String str) {
     if (str == null) {

diff --git a/api/src/test/java/io/kafbat/ui/KafkaConsumerGroupTests.java b/api/src/test/java/io/kafbat/ui/KafkaConsumerGroupTests.java
@@ -10,7 +10,6 @@
 import java.util.List;
 import java.util.Properties;
 import java.util.UUID;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import lombok.extern.slf4j.Slf4j;
 import lombok.val;
@@ -32,7 +31,6 @@ public class KafkaConsumerGroupTests extends AbstractIntegrationTest {
   @Test
   void shouldNotFoundWhenNoSuchConsumerGroupId() {
     String groupId = "groupA";
-    String expError = "The group id does not exist";
     webTestClient
         .delete()
         .uri("/api/clusters/{clusterName}/consumer-groups/{groupId}", LOCAL, groupId)
@@ -47,12 +45,13 @@ void shouldOkWhenConsumerGroupIsNotActive() {
 
     //Create a consumer and subscribe to the topic
     String groupId = UUID.randomUUID().toString();
-    val consumer = createTestConsumerWithGroupId(groupId);
-    consumer.subscribe(List.of(topicName));
-    consumer.poll(Duration.ofMillis(100));
+    try (val consumer = createTestConsumerWithGroupId(groupId)) {
+      consumer.subscribe(List.of(topicName));
+      consumer.poll(Duration.ofMillis(100));
 
-    //Unsubscribe from all topics to be able to delete this consumer
-    consumer.unsubscribe();
+      //Unsubscribe from all topics to be able to delete this consumer
+      consumer.unsubscribe();
+    }
 
     //Delete the consumer when it's INACTIVE and check
     webTestClient
@@ -69,24 +68,24 @@ void shouldBeBadRequestWhenConsumerGroupIsActive() {
 
     //Create a consumer and subscribe to the topic
     String groupId = UUID.randomUUID().toString();
-    val consumer = createTestConsumerWithGroupId(groupId);
-    consumer.subscribe(List.of(topicName));
-    consumer.poll(Duration.ofMillis(100));
+    try (val consumer = createTestConsumerWithGroupId(groupId)) {
+      consumer.subscribe(List.of(topicName));
+      consumer.poll(Duration.ofMillis(100));
 
-    //Try to delete the consumer when it's ACTIVE
-    String expError = "The group is not empty";
-    webTestClient
-        .delete()
-        .uri("/api/clusters/{clusterName}/consumer-groups/{groupId}", LOCAL, groupId)
-        .exchange()
-        .expectStatus()
-        .isBadRequest();
+      //Try to delete the consumer when it's ACTIVE
+      webTestClient
+          .delete()
+          .uri("/api/clusters/{clusterName}/consumer-groups/{groupId}", LOCAL, groupId)
+          .exchange()
+          .expectStatus()
+          .isBadRequest();
+    }
   }
 
   @Test
   void shouldReturnConsumerGroupsWithPagination() throws Exception {
-    try (var groups1 = startConsumerGroups(3, "cgPageTest1");
-        var groups2 = startConsumerGroups(2, "cgPageTest2")) {
+    try (var ignored = startConsumerGroups(3, "cgPageTest1");
+         var ignored1 = startConsumerGroups(2, "cgPageTest2")) {
       webTestClient
           .get()
           .uri("/api/clusters/{clusterName}/consumer-groups/paged?perPage=3&search=cgPageTest", LOCAL)
@@ -114,19 +113,19 @@ void shouldReturnConsumerGroupsWithPagination() throws Exception {
           });
 
       webTestClient
-            .get()
-            .uri("/api/clusters/{clusterName}/consumer-groups/paged?perPage=10&&search"
-                + "=cgPageTest&orderBy=NAME&sortOrder=DESC", LOCAL)
-            .exchange()
-            .expectStatus()
-            .isOk()
-            .expectBody(ConsumerGroupsPageResponseDTO.class)
-            .value(page -> {
-              assertThat(page.getPageCount()).isEqualTo(1);
-              assertThat(page.getConsumerGroups().size()).isEqualTo(5);
-              assertThat(page.getConsumerGroups())
-                  .isSortedAccordingTo(Comparator.comparing(ConsumerGroupDTO::getGroupId).reversed());
-            });
+          .get()
+          .uri("/api/clusters/{clusterName}/consumer-groups/paged?perPage=10&&search"
+              + "=cgPageTest&orderBy=NAME&sortOrder=DESC", LOCAL)
+          .exchange()
+          .expectStatus()
+          .isOk()
+          .expectBody(ConsumerGroupsPageResponseDTO.class)
+          .value(page -> {
+            assertThat(page.getPageCount()).isEqualTo(1);
+            assertThat(page.getConsumerGroups().size()).isEqualTo(5);
+            assertThat(page.getConsumerGroups())
+                .isSortedAccordingTo(Comparator.comparing(ConsumerGroupDTO::getGroupId).reversed());
+          });
 
       webTestClient
           .get()
@@ -156,7 +155,7 @@ private Closeable startConsumerGroups(int count, String consumerGroupPrefix) {
           return consumer;
         })
         .limit(count)
-        .collect(Collectors.toList());
+        .toList();
     return () -> {
       consumers.forEach(KafkaConsumer::close);
       deleteTopic(topicName);

diff --git a/api/src/test/java/io/kafbat/ui/emitter/MessageFiltersTest.java b/api/src/test/java/io/kafbat/ui/emitter/MessageFiltersTest.java
@@ -10,10 +10,11 @@
 import io.kafbat.ui.exception.CelException;
 import io.kafbat.ui.model.TopicMessageDTO;
 import java.time.OffsetDateTime;
-import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.function.Predicate;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.Nested;
@@ -100,7 +101,7 @@ void canCheckTimestampMs() {
       var ts = OffsetDateTime.now();
       var f = celScriptFilter("record.timestampMs == " + ts.toInstant().toEpochMilli());
       assertTrue(f.test(msg().timestamp(ts)));
-      assertFalse(f.test(msg().timestamp(ts.plus(1L, ChronoUnit.SECONDS))));
+      assertFalse(f.test(msg().timestamp(ts.plusSeconds(1L))));
     }
 
     @Test
@@ -177,6 +178,7 @@ void filterSpeedIsAtLeast5kPerSec() {
         toFilter.add(msg().content(jsonContent).key(randString));
       }
       // first iteration for warmup
+      // noinspection ResultOfMethodCallIgnored
       toFilter.stream().filter(f).count();
 
       long before = System.currentTimeMillis();
@@ -188,10 +190,15 @@ void filterSpeedIsAtLeast5kPerSec() {
     }
   }
 
+  @Test
+  void testBase64DecodingWorks() {
+    var uuid = UUID.randomUUID().toString();
+    var msg = "test." + Base64.getEncoder().encodeToString(uuid.getBytes());
+    var f = celScriptFilter("string(base64.decode(record.value.split('.')[1])).contains('" + uuid + "')");
+    assertTrue(f.test(msg().content(msg)));
+  }
+
   private TopicMessageDTO msg() {
-    return new TopicMessageDTO()
-        .timestamp(OffsetDateTime.now())
-        .offset(-1L)
-        .partition(1);
+    return new TopicMessageDTO(1, -1L, OffsetDateTime.now());
   }
 }
diff --git a/api/src/test/java/io/kafbat/ui/service/acl/AclCsvTest.java b/api/src/test/java/io/kafbat/ui/service/acl/AclCsvTest.java
@@ -6,6 +6,7 @@
 import io.kafbat.ui.exception.ValidationException;
 import java.util.Collection;
 import java.util.List;
+import java.util.stream.Stream;
 import org.apache.kafka.common.acl.AccessControlEntry;
 import org.apache.kafka.common.acl.AclBinding;
 import org.apache.kafka.common.acl.AclOperation;
@@ -15,6 +16,8 @@
 import org.apache.kafka.common.resource.ResourceType;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 class AclCsvTest {
@@ -29,22 +32,26 @@ class AclCsvTest {
   );
 
   @ParameterizedTest
-  @ValueSource(strings = {
-      "Principal,ResourceType, PatternType, ResourceName,Operation,PermissionType,Host\n"
-          + "User:test1,TOPIC,LITERAL,*,READ,ALLOW,*\n"
-          + "User:test2,GROUP,PREFIXED,group1,DESCRIBE,DENY,localhost",
-
-      //without header
-      "User:test1,TOPIC,LITERAL,*,READ,ALLOW,*\n"
-          + "\n"
-          + "User:test2,GROUP,PREFIXED,group1,DESCRIBE,DENY,localhost"
-          + "\n"
-  })
+  @MethodSource
   void parsesValidInputCsv(String csvString) {
     Collection<AclBinding> parsed = AclCsv.parseCsv(csvString);
     assertThat(parsed).containsExactlyInAnyOrderElementsOf(TEST_BINDINGS);
   }
 
+  private static Stream<Arguments> parsesValidInputCsv() {
+    return Stream.of(
+        Arguments.of(
+            "Principal,ResourceType, PatternType, ResourceName,Operation,PermissionType,Host" + System.lineSeparator()
+                + "User:test1,TOPIC,LITERAL,*,READ,ALLOW,*" + System.lineSeparator()
+                + "User:test2,GROUP,PREFIXED,group1,DESCRIBE,DENY,localhost"),
+        Arguments.of(
+            //without header
+            "User:test1,TOPIC,LITERAL,*,READ,ALLOW,*" + System.lineSeparator()
+                + System.lineSeparator()
+                + "User:test2,GROUP,PREFIXED,group1,DESCRIBE,DENY,localhost"
+                + System.lineSeparator()));
+  }
+
   @ParameterizedTest
   @ValueSource(strings = {
       // columns > 7

diff --git a/api/src/test/java/io/kafbat/ui/service/acl/AclsServiceTest.java b/api/src/test/java/io/kafbat/ui/service/acl/AclsServiceTest.java
@@ -68,8 +68,8 @@ void testSyncAclWithAclCsv() {
 
     aclsService.syncAclWithAclCsv(
         CLUSTER,
-        "Principal,ResourceType, PatternType, ResourceName,Operation,PermissionType,Host\n"
-            + "User:test1,TOPIC,LITERAL,*,READ,ALLOW,*\n"
+        "Principal,ResourceType, PatternType, ResourceName,Operation,PermissionType,Host" + System.lineSeparator()
+            + "User:test1,TOPIC,LITERAL,*,READ,ALLOW,*" + System.lineSeparator()
             + "User:test3,GROUP,PREFIXED,groupNew,DESCRIBE,DENY,localhost"
     ).block();
 

diff --git a/contract/src/main/resources/swagger/kafbat-ui-api.yaml b/contract/src/main/resources/swagger/kafbat-ui-api.yaml
@@ -3472,6 +3472,7 @@ components:
         - UNASSIGNED
         - TASK_FAILED
         - RESTARTING
+        - STOPPED
 
     ConnectorAction:
       type: string

diff --git a/contract/src/main/resources/swagger/kafka-connect-api.yaml b/contract/src/main/resources/swagger/kafka-connect-api.yaml
@@ -448,6 +448,7 @@ components:
                 - PAUSED
                 - UNASSIGNED
                 - RESTARTING
+                - STOPPED
             worker_id:
               type: string
             trace:

diff --git a/documentation/compose/DOCKER_COMPOSE.md b/documentation/compose/DOCKER_COMPOSE.md
@@ -2,15 +2,13 @@
 
 1. [kafka-ui.yaml](./kafbat-ui.yaml) - Default configuration with 2 kafka clusters with two nodes of Schema Registry, one kafka-connect and a few dummy topics.
 2. [kafka-ui-arm64.yaml](../../.dev/dev_arm64.yaml) - Default configuration for ARM64(Mac M1) architecture with 1 kafka cluster without zookeeper with one node of Schema Registry, one kafka-connect and a few dummy topics.
-3. [kafka-clusters-only.yaml](./kafka-clusters-only.yaml) - A configuration for development purposes, everything besides `kafka-ui` itself (to be run locally).
-4. [kafka-ui-ssl.yml](./kafka-ssl.yml) - Connect to Kafka via TLS/SSL
-5. [kafka-cluster-sr-auth.yaml](./cluster-sr-auth.yaml) - Schema registry with authentication.
-6. [kafka-ui-auth-context.yaml](./auth-context.yaml) - Basic (username/password) authentication with custom path (URL) (issue 861).
-7. [e2e-tests.yaml](./e2e-tests.yaml) - Configuration with different connectors (github-source, s3, sink-activities, source-activities) and Ksql functionality.
-8. [kafka-ui-jmx-secured.yml](./ui-jmx-secured.yml) - Kafka’s JMX with SSL and authentication.
-9. [kafka-ui-reverse-proxy.yaml](./nginx-proxy.yaml) - An example for using the app behind a proxy (like nginx).
-10. [kafka-ui-sasl.yaml](./ui-sasl.yaml) - SASL auth for Kafka.
-11. [kafka-ui-traefik-proxy.yaml](./traefik-proxy.yaml) - Traefik specific proxy configuration.
-12. [oauth-cognito.yaml](./oauth-cognito.yaml) - OAuth2 with Cognito
-13. [kafka-ui-with-jmx-exporter.yaml](./ui-with-jmx-exporter.yaml) - A configuration with 2 kafka clusters with enabled prometheus jmx exporters instead of jmx.
-14. [kafka-with-zookeeper.yaml](./kafka-zookeeper.yaml) - An example for using kafka with zookeeper
+3. [kafka-ui-ssl.yml](./kafka-ssl.yml) - Connect to Kafka via TLS/SSL
+4. [kafka-cluster-sr-auth.yaml](./cluster-sr-auth.yaml) - Schema registry with authentication.
+5. [kafka-ui-auth-context.yaml](./auth-context.yaml) - Basic (username/password) authentication with custom path (URL) (issue 861).
+6. [e2e-tests.yaml](./e2e-tests.yaml) - Configuration with different connectors (github-source, s3, sink-activities, source-activities) and Ksql functionality.
+7. [kafka-ui-jmx-secured.yml](./ui-jmx-secured.yml) - Kafka’s JMX with SSL and authentication.
+8. [kafka-ui-reverse-proxy.yaml](./nginx-proxy.yaml) - An example for using the app behind a proxy (like nginx).
+9. [kafka-ui-sasl.yaml](./ui-sasl.yaml) - SASL auth for Kafka.
+10. [kafka-ui-traefik-proxy.yaml](./traefik-proxy.yaml) - Traefik specific proxy configuration.
+11. [kafka-ui-with-jmx-exporter.yaml](./ui-with-jmx-exporter.yaml) - A configuration with 2 kafka clusters with enabled prometheus jmx exporters instead of jmx.
+12. [kafka-with-zookeeper.yaml](./kafka-zookeeper.yaml) - An example for using kafka with zookeeper
diff --git a/documentation/compose/ui-acl-with-zk.yaml b/documentation/compose/ui-acl-with-zk.yaml
@@ -18,7 +18,7 @@ services:
       KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";'
 
   zookeeper:
-    image: wurstmeister/zookeeper:3.4.6
+    image: zookeeper:3.8
     environment:
       JVMFLAGS: "-Djava.security.auth.login.config=/etc/zookeeper/zookeeper_jaas.conf"
     volumes:

diff --git a/frontend/src/components/common/Switch/Switch.styled.ts b/frontend/src/components/common/Switch/Switch.styled.ts
@@ -1,13 +1,13 @@
 import styled from 'styled-components';
 
 interface Props {
-  isCheckedIcon?: boolean;
+  checked?: boolean;
 }
 
-export const StyledLabel = styled.label<Props>`
+export const StyledLabel = styled.label`
   position: relative;
   display: inline-block;
-  width: ${({ isCheckedIcon }) => (isCheckedIcon ? '40px' : '34px')};
+  width: 34px;
   height: 20px;
   margin-right: 8px;
 `;
@@ -32,14 +32,12 @@ export const StyledSlider = styled.span<Props>`
   left: 0;
   right: 0;
   bottom: 0;
-  background-color: ${({ isCheckedIcon, theme }) =>
-    isCheckedIcon
-      ? theme.switch.checkedIcon.backgroundColor
-      : theme.switch.unchecked};
+  background-color: ${({ checked, theme }) =>
+    checked ? theme.switch.checked : theme.switch.unchecked};
   transition: 0.4s;
   border-radius: 20px;
 
-  :hover {
+  &:hover {
     background-color: ${({ theme }) => theme.switch.hover};
   }
 
@@ -48,7 +46,7 @@ export const StyledSlider = styled.span<Props>`
     content: '';
     height: 14px;
     width: 14px;
-    left: 3px;
+    left: ${({ checked }) => (checked ? '17px' : '3px')};
     bottom: 3px;
     background-color: ${({ theme }) => theme.switch.circle};
     transition: 0.4s;
@@ -57,25 +55,8 @@ export const StyledSlider = styled.span<Props>`
   }
 `;
 
-export const StyledInput = styled.input<Props>`
+export const StyledInput = styled.input`
   opacity: 0;
   width: 0;
   height: 0;
-
-  &:checked + ${StyledSlider} {
-    background-color: ${({ isCheckedIcon, theme }) =>
-      isCheckedIcon
-        ? theme.switch.checkedIcon.backgroundColor
-        : theme.switch.checked};
-  }
-
-  &:focus + ${StyledSlider} {
-    box-shadow: 0 0 1px ${({ theme }) => theme.switch.checked};
-  }
-
-  :checked + ${StyledSlider}:before {
-    transform: translateX(
-      ${({ isCheckedIcon }) => (isCheckedIcon ? '20px' : '14px')}
-    );
-  }
 `;