Merge branch 'main' into feature/reset-connector-offsets

kafbat · Jan 13, 2025 · 513e3f9 · 513e3f9
2 parents fbb6ddd + 4cf17a0
commit 513e3f9
Show file tree

Hide file tree

Showing 21 changed files with 227 additions and 32 deletions.
diff --git a/.github/workflows/backend_main.yml b/.github/workflows/backend_main.yml
@@ -19,7 +19,7 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  build:
+  build-and-test:
     uses: ./.github/workflows/backend_tests.yml
     with:
       event_name: ${{ github.event_name }}
diff --git a/.github/workflows/backend_pr.yml b/.github/workflows/backend_pr.yml
@@ -20,7 +20,7 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  build:
+  build-and-test:
     uses: ./.github/workflows/backend_tests.yml
     with:
       event_name: ${{ github.event_name }}
diff --git a/.github/workflows/cve_checks.yml b/.github/workflows/cve_checks.yml
@@ -9,7 +9,8 @@ permissions:
   contents: read
 
 jobs:
-  build-and-test:
+
+  check-cves:
     runs-on: ubuntu-latest
 
     steps:
@@ -67,3 +68,12 @@ jobs:
           image-ref: "ghcr.io/kafbat/kafka-ui:${{ steps.build.outputs.version }}"
           format: "table"
           exit-code: "1"
+
+  notify:
+    needs: check-cves
+    if: ${{ always() && needs.build-and-test.result == 'failure' }}
+    uses: ./.github/workflows/infra_discord_hook.yml
+    with:
+      message: "Attention! CVE checks run failed! Please fix them CVEs :("
+    secrets:
+      DISCORD_WEBHOOK_URL: ${{ secrets.DISCORD_WEBHOOK_URL_CVE }}
diff --git a/.github/workflows/frontend_main.yml b/.github/workflows/frontend_main.yml
@@ -15,5 +15,5 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  build:
+  build-and-test:
     uses: ./.github/workflows/frontend_tests.yml
diff --git a/.github/workflows/frontend_pr.yml b/.github/workflows/frontend_pr.yml
@@ -16,5 +16,5 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  build:
+  build-and-test:
     uses: ./.github/workflows/frontend_tests.yml
diff --git a/.github/workflows/infra_discord_hook.yml b/.github/workflows/infra_discord_hook.yml
@@ -0,0 +1,27 @@
+name: 'Discord hook'
+
+on:
+  workflow_call:
+    inputs:
+      message:
+        description: 'Message text'
+        required: true
+        type: string
+    secrets:
+      DISCORD_WEBHOOK_URL:
+        required: true
+
+permissions:
+  contents: read
+
+jobs:
+
+  hook:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Notify Discord on Failure
+        uses: Ilshidur/[email protected]
+        with:
+          args: ${{ inputs.message }}
+        env:
+          DISCORD_WEBHOOK: ${{ secrets.DISCORD_WEBHOOK_URL }}
diff --git a/.github/workflows/md-links.yml b/.github/workflows/md-links.yml
@@ -13,7 +13,7 @@ permissions:
   contents: read
 
 jobs:
-  build-and-test:
+  lint-md:
     runs-on: ubuntu-latest
 
     steps:

diff --git a/.github/workflows/pr_linter.yml b/.github/workflows/pr_linter.yml
@@ -5,13 +5,12 @@ on:
 permissions:
   checks: write
 jobs:
-  task-check:
+  check-tasks:
     runs-on: ubuntu-latest
     steps:
       - uses: kentaro-m/[email protected]
         with:
           repo-token: "${{ secrets.GITHUB_TOKEN }}"
       - uses: dekinderfiets/[email protected]
-        if: false # TODO remove when public
         with:
           repo-token: "${{ secrets.GITHUB_TOKEN }}"
diff --git a/.github/workflows/workflow_linter.yml b/.github/workflows/workflow_linter.yml
@@ -9,7 +9,7 @@ permissions:
   contents: read
 
 jobs:
-  build-and-test:
+  lint-workflows:
     runs-on: ubuntu-latest
 
     steps:

diff --git a/api/src/main/java/io/kafbat/ui/config/CorsGlobalConfiguration.java b/api/src/main/java/io/kafbat/ui/config/CorsGlobalConfiguration.java
@@ -22,10 +22,7 @@ public WebFilter corsFilter() {
 
       final ServerHttpResponse response = ctx.getResponse();
       final HttpHeaders headers = response.getHeaders();
-      headers.add("Access-Control-Allow-Origin", "*");
-      headers.add("Access-Control-Allow-Methods", "GET, PUT, POST, DELETE, OPTIONS");
-      headers.add("Access-Control-Max-Age", "3600");
-      headers.add("Access-Control-Allow-Headers", "Content-Type");
+      fillCorsHeader(headers, request);
 
       if (request.getMethod() == HttpMethod.OPTIONS) {
         response.setStatusCode(HttpStatus.OK);
@@ -36,4 +33,11 @@ public WebFilter corsFilter() {
     };
   }
 
+  public static void fillCorsHeader(HttpHeaders responseHeaders, ServerHttpRequest request) {
+    responseHeaders.add("Access-Control-Allow-Origin", request.getHeaders().getOrigin());
+    responseHeaders.add("Access-Control-Allow-Credentials", "true");
+    responseHeaders.add("Access-Control-Allow-Methods", "GET, PUT, POST, DELETE, OPTIONS");
+    responseHeaders.add("Access-Control-Max-Age", "3600");
+    responseHeaders.add("Access-Control-Allow-Headers", "Content-Type");
+  }
 }
diff --git a/api/src/main/java/io/kafbat/ui/config/ReadOnlyModeFilter.java b/api/src/main/java/io/kafbat/ui/config/ReadOnlyModeFilter.java
@@ -33,7 +33,8 @@ public class ReadOnlyModeFilter implements WebFilter {
   @NotNull
   @Override
   public Mono<Void> filter(ServerWebExchange exchange, @NotNull WebFilterChain chain) {
-    var isSafeMethod = exchange.getRequest().getMethod() == HttpMethod.GET;
+    var isSafeMethod =
+        exchange.getRequest().getMethod() == HttpMethod.GET || exchange.getRequest().getMethod() == HttpMethod.OPTIONS;
     if (isSafeMethod) {
       return chain.filter(exchange);
     }

diff --git a/api/src/main/java/io/kafbat/ui/exception/GlobalErrorWebExceptionHandler.java b/api/src/main/java/io/kafbat/ui/exception/GlobalErrorWebExceptionHandler.java
@@ -2,12 +2,14 @@
 
 import com.google.common.base.Throwables;
 import com.google.common.collect.Sets;
+import io.kafbat.ui.config.CorsGlobalConfiguration;
 import io.kafbat.ui.model.ErrorResponseDTO;
 import java.math.BigDecimal;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.springframework.boot.autoconfigure.web.WebProperties;
@@ -16,6 +18,7 @@
 import org.springframework.context.ApplicationContext;
 import org.springframework.core.Ordered;
 import org.springframework.core.annotation.Order;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.ServerCodecConfigurer;
@@ -78,6 +81,7 @@ private Mono<ServerResponse> renderDefault(Throwable throwable, ServerRequest re
     return ServerResponse
         .status(ErrorCode.UNEXPECTED.httpStatus())
         .contentType(MediaType.APPLICATION_JSON)
+        .headers(headers(request))
         .bodyValue(response);
   }
 
@@ -92,6 +96,7 @@ private Mono<ServerResponse> render(CustomBaseException baseException, ServerReq
     return ServerResponse
         .status(errorCode.httpStatus())
         .contentType(MediaType.APPLICATION_JSON)
+        .headers(headers(request))
         .bodyValue(response);
   }
 
@@ -122,6 +127,7 @@ private Mono<ServerResponse> render(WebExchangeBindException exception, ServerRe
     return ServerResponse
         .status(HttpStatus.BAD_REQUEST)
         .contentType(MediaType.APPLICATION_JSON)
+        .headers(headers(request))
         .bodyValue(response);
   }
 
@@ -136,13 +142,20 @@ private Mono<ServerResponse> render(ResponseStatusException exception, ServerReq
     return ServerResponse
         .status(exception.getStatusCode())
         .contentType(MediaType.APPLICATION_JSON)
+        .headers(headers(request))
         .bodyValue(response);
   }
 
   private String requestId(ServerRequest request) {
     return request.exchange().getRequest().getId();
   }
 
+  private Consumer<HttpHeaders> headers(ServerRequest request) {
+    return (HttpHeaders headers) -> {
+      CorsGlobalConfiguration.fillCorsHeader(headers, request.exchange().getRequest());
+    };
+  }
+
   private BigDecimal currentTimestamp() {
     return BigDecimal.valueOf(System.currentTimeMillis());
   }

diff --git a/api/src/main/java/io/kafbat/ui/util/StaticFileWebFilter.java b/api/src/main/java/io/kafbat/ui/util/StaticFileWebFilter.java
@@ -1,6 +1,7 @@
 package io.kafbat.ui.util;
 
 import java.io.IOException;
+import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.buffer.DataBufferFactory;
@@ -15,6 +16,7 @@
 import org.springframework.web.server.WebFilterChain;
 import reactor.core.publisher.Mono;
 
+@Slf4j
 public class StaticFileWebFilter implements WebFilter {
 
   private static final String INDEX_HTML = "/static/index.html";
@@ -29,6 +31,12 @@ public StaticFileWebFilter() {
   public StaticFileWebFilter(String path, ClassPathResource resource) {
     this.matcher = ServerWebExchangeMatchers.pathMatchers(HttpMethod.GET, path);
 
+    if (!resource.exists()) {
+      log.warn("Resource [{}] does not exist. Frontend might not be available.", resource.getPath());
+      contents = "Missing index.html. Make sure the app has been built with a correct (prod) profile.";
+      return;
+    }
+
     try {
       this.contents = ResourceUtil.readAsString(resource);
     } catch (IOException e) {

diff --git a/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java b/api/src/main/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverter.java
@@ -80,14 +80,12 @@ private FieldSchema createUnionSchema(Schema schema, Map<String, FieldSchema> de
     final Map<String, FieldSchema> fields = schema.getTypes().stream()
         .filter(t -> !t.getType().equals(Schema.Type.NULL))
         .map(f -> {
-          String oneOfFieldName;
-          if (f.getType().equals(Schema.Type.RECORD)) {
-            // for records using full record name
-            oneOfFieldName = f.getFullName();
-          } else {
+          String oneOfFieldName = switch (f.getType()) {
+            case RECORD -> f.getFullName();
+            case ENUM -> f.getName();
             // for primitive types - using type name
-            oneOfFieldName = f.getType().getName().toLowerCase();
-          }
+            default -> f.getType().getName().toLowerCase();
+          };
           return Tuples.of(oneOfFieldName, convertSchema(f, definitions, false));
         }).collect(Collectors.toMap(
             Tuple2::getT1,

diff --git a/api/src/test/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverterTest.java b/api/src/test/java/io/kafbat/ui/util/jsonschema/AvroJsonSchemaConverterTest.java
@@ -244,6 +244,48 @@ void testRecordReferences() {
     convertAndCompare(expectedJsonSchema, avroSchema);
   }
 
+  @Test
+  void testNullableUnionEnum()  {
+    String avroSchema =
+        " {"
+            + "     \"type\": \"record\","
+            + "     \"name\": \"Message\","
+            + "     \"namespace\": \"com.provectus.kafka\","
+            + "     \"fields\": ["
+            + "                     {"
+            + "                         \"name\": \"enum_nullable_union\","
+            + "                         \"type\": [\"null\", {"
+            + "                             \"type\": \"enum\","
+            + "                             \"name\": \"Suit\","
+            + "                             \"symbols\": [\"SPADES\",\"HEARTS\",\"DIAMONDS\",\"CLUBS\"]"
+            + "                         }]"
+            + "                     }"
+            + "     ]"
+            + " }";
+
+    String expectedJsonSchema =
+        "{\"$id\":\"http://example.com/Message\","
+            + "\"$schema\":\"https://json-schema.org/draft/2020-12/schema\","
+            + "\"type\":\"object\","
+            + "\"properties\":{"
+            + "\"enum_nullable_union\":{"
+            + "\"oneOf\":["
+            + "{\"type\":\"null\"},"
+            + "{\"type\":\"object\","
+            + "\"properties\":{"
+            + "\"Suit\":{"
+            + "\"type\":\"string\","
+            + "\"enum\":[\"SPADES\",\"HEARTS\",\"DIAMONDS\",\"CLUBS\"]"
+            + "}}}"
+            + "]"
+            + "}},"
+            + "\"definitions\":{"
+            + "\"com.provectus.kafka.Message\":{\"$ref\":\"#\"}"
+            + "}}";
+
+    convertAndCompare(expectedJsonSchema, avroSchema);
+  }
+
   @SneakyThrows
   private void convertAndCompare(String expectedJsonSchema, String sourceAvroSchema) {
     var parseAvroSchema = new Schema.Parser().parse(sourceAvroSchema);

diff --git a/api/src/test/java/io/kafbat/ui/util/jsonschema/JsonAvroConversionTest.java b/api/src/test/java/io/kafbat/ui/util/jsonschema/JsonAvroConversionTest.java
@@ -700,6 +700,43 @@ void unionFieldWithInnerTypesNamesClash() {
 
   }
 
+  @Test
+  void unionNullableEnumField() {
+    var schema = createSchema(
+        """
+             {
+               "type": "record",
+               "namespace": "com.test",
+               "name": "TestAvroRecord",
+               "fields": [
+                 {
+                   "name": "enum_nullable_union",
+                   "type" : [ "null", {
+                               "type" : "enum",
+                               "name" : "Suit",
+                               "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
+                             } ]
+                 }
+               ]
+            }"""
+    );
+
+    GenericData.Record inputRecord = new GenericData.Record(schema);
+    inputRecord.put("enum_nullable_union",
+        new GenericData.EnumSymbol(
+            schema.getField("enum_nullable_union").schema().getTypes().get(1), "SPADES"));
+    String expectedJsonWithEnum = """
+          {
+            "enum_nullable_union": { "Suit": "SPADES"}\s
+          }
+         \s""";
+    assertJsonsEqual(expectedJsonWithEnum, convertAvroToJson(inputRecord, schema));
+
+    GenericData.Record inputNullRecord  = new GenericData.Record(schema);
+    inputNullRecord.put("enum_nullable_union", null);
+    assertJsonsEqual("{}", convertAvroToJson(inputNullRecord, schema));
+  }
+
   private Schema createSchema(String schema) {
     return new AvroSchema(schema).rawSchema();
   }

diff --git a/frontend/package.json b/frontend/package.json
@@ -104,7 +104,7 @@
   },
   "engines": {
     "node": "18.17.1",
-    "pnpm": "9.15.0"
+    "pnpm": "^9.15.0"
   },
   "pnpm": {
     "overrides": {
@@ -117,6 +117,7 @@
       "json5@>=2.0.0 <2.2.2": ">=2.2.2",
       "semver@>=7.0.0 <7.5.2": ">=7.5.2",
       "axios@>=0.8.1 <0.28.0": ">=0.28.0",
+      "axios@>=1.3.2 <=1.7.3": ">=1.7.4",
       "braces": "3.0.3"
     }
   }