Fix rebase issues

kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilters.java

@@ -31,41 +31,42 @@ public class MessageFilters {
   }
 
   public static Predicate<TopicMessageDTO> groovyScriptFilter(String script) {
-    var compiledScript = compileScript(script);
+    var engine = getGroovyEngine();
+    var compiledScript = compileScript(engine, script);
     var jsonSlurper = new JsonSlurper();
     return new Predicate<TopicMessageDTO>() {
       @SneakyThrows
       @Override
       public boolean test(TopicMessageDTO msg) {
-        var bindings = getGroovyEngine().createBindings();
+        var bindings = engine.createBindings();
         bindings.put("partition", msg.getPartition());
         bindings.put("offset", msg.getOffset());
         bindings.put("timestampMs", msg.getTimestamp().toInstant().toEpochMilli());
         bindings.put("keyAsText", msg.getKey());
         bindings.put("valueAsText", msg.getContent());
         bindings.put("headers", msg.getHeaders());
-        bindings.put("key", parseToJsonOrReturnNull(jsonSlurper, msg.getKey()));
-        bindings.put("value", parseToJsonOrReturnNull(jsonSlurper, msg.getContent()));
+        bindings.put("key", parseToJsonOrReturnAsIs(jsonSlurper, msg.getKey()));
+        bindings.put("value", parseToJsonOrReturnAsIs(jsonSlurper, msg.getContent()));
         var result = compiledScript.eval(bindings);
         if (result instanceof Boolean) {
           return (Boolean) result;
         } else {
           throw new ValidationException(
-              String.format("Unexpected script result: %s, Boolean should be returned instead", result));
+              "Unexpected script result: %s, Boolean should be returned instead".formatted(result));
         }
       }
     };
   }
 
   @Nullable
-  private static Object parseToJsonOrReturnNull(JsonSlurper parser, @Nullable String str) {
+  private static Object parseToJsonOrReturnAsIs(JsonSlurper parser, @Nullable String str) {
     if (str == null) {
       return null;
     }
     try {
       return parser.parseText(str);
     } catch (Exception e) {
-      return null;
+      return str;
     }
   }
 
@@ -78,9 +79,9 @@ public class MessageFilters {
     return GROOVY_ENGINE;
   }
 
-  private static CompiledScript compileScript(String script) {
+  private static CompiledScript compileScript(GroovyScriptEngineImpl engine, String script) {
     try {
-      return getGroovyEngine().compile(script);
+      return engine.compile(script);
     } catch (ScriptException e) {
       throw new ValidationException("Script syntax error: " + e.getMessage());
     }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/OffsetsInfo.java

@@ -9,6 +9,7 @@ import java.util.Set;
 import java.util.stream.Collectors;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.mutable.MutableLong;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
 
@@ -57,6 +58,12 @@ public class OffsetsInfo {
     return true;
   }
 
+  public long summaryOffsetsRange() {
+    MutableLong cnt = new MutableLong();
+    nonEmptyPartitions.forEach(tp -> cnt.add(endOffsets.get(tp) - beginOffsets.get(tp)));
+    return cnt.getValue();
+  }
+
   public Set<TopicPartition> allTargetPartitions() {
     return Sets.union(nonEmptyPartitions, emptyPartitions);
   }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/SeekOperations.java

@@ -10,11 +10,12 @@ import java.util.HashMap;
 import java.util.Map;
 import lombok.AccessLevel;
 import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.mutable.MutableLong;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
 
 @RequiredArgsConstructor(access = AccessLevel.PACKAGE)
-class SeekOperations {
+public class SeekOperations {
 
   private final Consumer<?, ?> consumer;
   private final OffsetsInfo offsetsInfo;
@@ -33,16 +34,32 @@ class SeekOperations {
     offsetsForSeek.forEach(consumer::seek);
   }
 
-  Map<TopicPartition, Long> getBeginOffsets() {
+  public Map<TopicPartition, Long> getBeginOffsets() {
     return offsetsInfo.getBeginOffsets();
   }
 
-  boolean assignedPartitionsFullyPolled() {
+  public Map<TopicPartition, Long> getEndOffsets() {
+    return offsetsInfo.getEndOffsets();
+  }
+
+  public boolean assignedPartitionsFullyPolled() {
     return offsetsInfo.assignedPartitionsFullyPolled();
   }
 
+  // sum of (end - start) offsets for all partitions
+  public long summaryOffsetsRange() {
+    return offsetsInfo.summaryOffsetsRange();
+  }
+
+  // sum of differences between initial consumer seek and current consumer position (across all partitions)
+  public long offsetsProcessedFromSeek() {
+    MutableLong count = new MutableLong();
+    offsetsForSeek.forEach((tp, initialOffset) -> count.add(consumer.position(tp) - initialOffset));
+    return count.getValue();
+  }
+
   // Get offsets to seek to. NOTE: offsets do not contain empty partitions offsets
-  Map<TopicPartition, Long> getOffsetsForSeek() {
+  public Map<TopicPartition, Long> getOffsetsForSeek() {
     return offsetsForSeek;
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/ConsumerRecordDeserializer.java

@@ -1,6 +1,7 @@
 package com.provectus.kafka.ui.serdes;
 
 import com.provectus.kafka.ui.model.TopicMessageDTO;
+import com.provectus.kafka.ui.model.TopicMessageDTO.TimestampTypeEnum;
 import com.provectus.kafka.ui.serde.api.Serde;
 import java.time.Instant;
 import java.time.OffsetDateTime;
@@ -8,6 +9,7 @@ import java.time.ZoneId;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.function.UnaryOperator;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -32,6 +34,8 @@ public class ConsumerRecordDeserializer {
   private final Serde.Deserializer fallbackKeyDeserializer;
   private final Serde.Deserializer fallbackValueDeserializer;
 
+  private final UnaryOperator<TopicMessageDTO> masker;
+
   public TopicMessageDTO deserialize(ConsumerRecord<Bytes, Bytes> rec) {
     var message = new TopicMessageDTO();
     fillKey(message, rec);
@@ -47,14 +51,14 @@ public class ConsumerRecordDeserializer {
     message.setValueSize(getValueSize(rec));
     message.setHeadersSize(getHeadersSize(rec));
 
-    return message;
+    return masker.apply(message);
   }
 
-  private static TopicMessageDTO.TimestampTypeEnum mapToTimestampType(TimestampType timestampType) {
+  private static TimestampTypeEnum mapToTimestampType(TimestampType timestampType) {
     return switch (timestampType) {
-      case CREATE_TIME -> TopicMessageDTO.TimestampTypeEnum.CREATE_TIME;
-      case LOG_APPEND_TIME -> TopicMessageDTO.TimestampTypeEnum.LOG_APPEND_TIME;
-      case NO_TIMESTAMP_TYPE -> TopicMessageDTO.TimestampTypeEnum.NO_TIMESTAMP_TYPE;
+      case CREATE_TIME -> TimestampTypeEnum.CREATE_TIME;
+      case LOG_APPEND_TIME -> TimestampTypeEnum.LOG_APPEND_TIME;
+      case NO_TIMESTAMP_TYPE -> TimestampTypeEnum.NO_TIMESTAMP_TYPE;
     };
   }
 
@@ -118,11 +122,11 @@ public class ConsumerRecordDeserializer {
   }
 
   private static Long getKeySize(ConsumerRecord<Bytes, Bytes> consumerRecord) {
-    return consumerRecord.key() != null ? (long) consumerRecord.key().get().length : null;
+    return consumerRecord.key() != null ? (long) consumerRecord.serializedKeySize() : null;
   }
 
   private static Long getValueSize(ConsumerRecord<Bytes, Bytes> consumerRecord) {
-    return consumerRecord.value() != null ? (long) consumerRecord.value().get().length : null;
+    return consumerRecord.value() != null ? (long) consumerRecord.serializedValueSize() : null;
   }
 
   private static int headerSize(Header header) {

kafka-ui-api/src/main/resources/application-local.yml

@@ -10,22 +10,22 @@ logging:
 #server:
 #  port: 8080 #- Port in which kafka-ui will run.
 
-#spring:
-#  jmx:
-#    enabled: true
-#  ldap:
-#    urls: ldap://localhost:10389
-#    base: "cn={0},ou=people,dc=planetexpress,dc=com"
-#    admin-user: "cn=admin,dc=planetexpress,dc=com"
-#    admin-password: "GoodNewsEveryone"
-#    user-filter-search-base: "dc=planetexpress,dc=com"
-#    user-filter-search-filter: "(&(uid={0})(objectClass=inetOrgPerson))"
-#    group-filter-search-base: "ou=people,dc=planetexpress,dc=com"
+spring:
+  jmx:
+    enabled: true
+  ldap:
+    urls: ldap://localhost:10389
+    base: "cn={0},ou=people,dc=planetexpress,dc=com"
+    admin-user: "cn=admin,dc=planetexpress,dc=com"
+    admin-password: "GoodNewsEveryone"
+    user-filter-search-base: "dc=planetexpress,dc=com"
+    user-filter-search-filter: "(&(uid={0})(objectClass=inetOrgPerson))"
+    group-filter-search-base: "ou=people,dc=planetexpress,dc=com"
 
 kafka:
   clusters:
     - name: local
-      bootstrapServers: localhost:9096
+      bootstrapServers: localhost:9092
       schemaRegistry: http://localhost:8085
       ksqldbServer: http://localhost:8088
       kafkaConnect:
@@ -80,70 +80,70 @@ auth:
         custom-params:
           type: github
 
-#rbac:
-#  roles:
-#    - name: "memelords"
-#      clusters:
-#        - local
-#      subjects:
-#        - provider: oauth_google
-#          type: domain
-#          value: "provectus.com"
-#        - provider: oauth_google
-#          type: user
-#          value: "name@provectus.com"
-#
-#        - provider: oauth_github
-#          type: organization
-#          value: "provectus"
-#        - provider: oauth_github
-#          type: user
-#          value: "memelord"
-#
-#        - provider: oauth_cognito
-#          type: user
-#          value: "username"
-#        - provider: oauth_cognito
-#          type: group
-#          value: "memelords"
-#
-#        - provider: ldap
-#          type: group
-#          value: "admin_staff"
-#
-#        # NOT IMPLEMENTED YET
-#      #        - provider: ldap_ad
-#      #          type: group
-#      #          value: "admin_staff"
-#
-#      permissions:
-#        - resource: applicationconfig
-#          actions: all
-#
-#        - resource: clusterconfig
-#          actions: all
-#
-#        - resource: topic
-#          value: ".*"
-#          actions: all
-#
-#        - resource: consumer
-#          value: ".*"
-#          actions: all
-#
-#        - resource: schema
-#          value: ".*"
-#          actions: all
-#
-#        - resource: connect
-#          value: "*"
-#          actions: all
-#
-#        - resource: ksql
-#          actions: all
-#
-#        - resource: acl
-#          actions: all
-#
-#        - resource: audit
-#          actions: all
+rbac:
+  roles:
+    - name: "memelords"
+      clusters:
+        - local
+      subjects:
+        - provider: oauth_google
+          type: domain
+          value: "provectus.com"
+        - provider: oauth_google
+          type: user
+          value: "name@provectus.com"
+
+        - provider: oauth_github
+          type: organization
+          value: "provectus"
+        - provider: oauth_github
+          type: user
+          value: "memelord"
+
+        - provider: oauth_cognito
+          type: user
+          value: "username"
+        - provider: oauth_cognito
+          type: group
+          value: "memelords"
+
+        - provider: ldap
+          type: group
+          value: "admin_staff"
+
+        # NOT IMPLEMENTED YET
+      #        - provider: ldap_ad
+      #          type: group
+      #          value: "admin_staff"
+
+      permissions:
+        - resource: applicationconfig
+          actions: all
+
+        - resource: clusterconfig
+          actions: all
+
+        - resource: topic
+          value: ".*"
+          actions: all
+
+        - resource: consumer
+          value: ".*"
+          actions: all
+
+        - resource: schema
+          value: ".*"
+          actions: all
+
+        - resource: connect
+          value: "*"
+          actions: all
+
+        - resource: ksql
+          actions: all
+
+        - resource: acl
+          actions: all
+
+        - resource: audit
+          actions: all

kafka-ui-api/src/main/resources/application.yml

@@ -1,109 +1,20 @@
+auth:
+  type: DISABLED
+
+management:
+  endpoint:
+    info:
+      enabled: true
+    health:
+      enabled: true
+  endpoints:
+    web:
+      exposure:
+        include: "info,health,prometheus"
+
 logging:
   level:
     root: INFO
     com.provectus: DEBUG
-    #org.springframework.http.codec.json.Jackson2JsonEncoder: DEBUG
-    #org.springframework.http.codec.json.Jackson2JsonDecoder: DEBUG
     reactor.netty.http.server.AccessLog: INFO
-    org.springframework.security: DEBUG
-
-#server:
-#  port: 8080 #- Port in which kafka-ui will run.
-
-#spring:
-#  jmx:
-#    enabled: true
-#  ldap:
-#    urls: ldap://localhost:10389
-#    base: "cn={0},ou=people,dc=planetexpress,dc=com"
-#    admin-user: "cn=admin,dc=planetexpress,dc=com"
-#    admin-password: "GoodNewsEveryone"
-#    user-filter-search-base: "dc=planetexpress,dc=com"
-#    user-filter-search-filter: "(&(uid={0})(objectClass=inetOrgPerson))"
-#    group-filter-search-base: "ou=people,dc=planetexpress,dc=com"
-
-kafka:
-  clusters:
-    - name: local
-      bootstrapServers: localhost:9096
-#      schemaRegistry: http://localhost:8085
-#      ksqldbServer: http://localhost:8088
-#      kafkaConnect:
-#        - name: first
-#          address: http://localhost:8083
-#      metrics:
-#        port: 9997
-#        type: JMX
-
-auth:
-  type: DISABLED
-
-dynamic.config.enabled: true
-
-#rbac:
-#  roles:
-#    - name: "memelords"
-#      clusters:
-#        - local
-#      subjects:
-#        - provider: oauth_google
-#          type: domain
-#          value: "provectus.com"
-#        - provider: oauth_google
-#          type: user
-#          value: "name@provectus.com"
-#
-#        - provider: oauth_github
-#          type: organization
-#          value: "provectus"
-#        - provider: oauth_github
-#          type: user
-#          value: "memelord"
-#
-#        - provider: oauth_cognito
-#          type: user
-#          value: "username"
-#        - provider: oauth_cognito
-#          type: group
-#          value: "memelords"
-#
-#        - provider: ldap
-#          type: group
-#          value: "admin_staff"
-#
-#        # NOT IMPLEMENTED YET
-#      #        - provider: ldap_ad
-#      #          type: group
-#      #          value: "admin_staff"
-#
-#      permissions:
-#        - resource: applicationconfig
-#          actions: all
-#
-#        - resource: clusterconfig
-#          actions: all
-#
-#        - resource: topic
-#          value: ".*"
-#          actions: all
-#
-#        - resource: consumer
-#          value: ".*"
-#          actions: all
-#
-#        - resource: schema
-#          value: ".*"
-#          actions: all
-#
-#        - resource: connect
-#          value: "*"
-#          actions: all
-#
-#        - resource: ksql
-#          actions: all
-#
-#        - resource: acl
-#          actions: all
-#
-#        - resource: audit
-#          actions: all
+    org.hibernate.validator: WARN

kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/MessagesServiceTest.java

@@ -162,4 +162,40 @@ class MessagesServiceTest extends AbstractIntegrationTest {
     createdTopics.add(newTopic.name());
   }
 
+  @Test
+  void execSmartFilterTestReturnsExecutionResult() {
+    var params = new SmartFilterTestExecutionDTO()
+        .filterCode("key != null && value != null && headers != null && timestampMs != null && offset != null")
+        .key("1234")
+        .value("{ \"some\" : \"value\" } ")
+        .headers(Map.of("h1", "hv1"))
+        .offset(12345L)
+        .timestampMs(System.currentTimeMillis())
+        .partition(1);
+    assertThat(execSmartFilterTest(params).getResult()).isTrue();
+
+    params.setFilterCode("return false");
+    assertThat(execSmartFilterTest(params).getResult()).isFalse();
+  }
+
+  @Test
+  void execSmartFilterTestReturnsErrorOnFilterApplyError() {
+    var result = execSmartFilterTest(
+        new SmartFilterTestExecutionDTO()
+            .filterCode("return 1/0")
+    );
+    assertThat(result.getResult()).isNull();
+    assertThat(result.getError()).containsIgnoringCase("execution error");
+  }
+
+  @Test
+  void execSmartFilterTestReturnsErrorOnFilterCompilationError() {
+    var result = execSmartFilterTest(
+        new SmartFilterTestExecutionDTO()
+            .filterCode("this is invalid groovy syntax = 1")
+    );
+    assertThat(result.getResult()).isNull();
+    assertThat(result.getError()).containsIgnoringCase("Compilation error");
+  }
+
 }

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -625,6 +625,25 @@ paths:
               schema:
                 $ref: '#/components/schemas/TopicSerdeSuggestion'
 
+  /api/smartfilters/testexecutions:
+    put:
+      tags:
+        - Messages
+      summary: executeSmartFilterTest
+      operationId: executeSmartFilterTest
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/SmartFilterTestExecution'
+      responses:
+        200:
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/SmartFilterTestExecutionResult'
+
 
   /api/clusters/{clusterName}/topics/{topicName}/messages:
     get:
@@ -1870,6 +1889,188 @@ paths:
         404:
           description: Not found
 
+  /api/clusters/{clusterName}/acls:
+    get:
+      tags:
+        - Acls
+      summary: listKafkaAcls
+      operationId: listAcls
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: resourceType
+          in: query
+          required: false
+          schema:
+            $ref: '#/components/schemas/KafkaAclResourceType'
+        - name: resourceName
+          in: query
+          required: false
+          schema:
+            type: string
+        - name: namePatternType
+          in: query
+          required: false
+          schema:
+            $ref: '#/components/schemas/KafkaAclNamePatternType'
+      responses:
+        200:
+          description: OK
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/KafkaAcl'
+
+  /api/clusters/{clusterName}/acl/csv:
+    get:
+      tags:
+        - Acls
+      summary: getAclAsCsv
+      operationId: getAclAsCsv
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        200:
+          description: OK
+          content:
+            text/plain:
+              schema:
+                type: string
+    post:
+      tags:
+        - Acls
+      summary: syncAclsCsv
+      operationId: syncAclsCsv
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          text/plain:
+            schema:
+              type: string
+      responses:
+        200:
+          description: OK
+
+  /api/clusters/{clusterName}/acl:
+    post:
+      tags:
+        - Acls
+      summary: createAcl
+      operationId: createAcl
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/KafkaAcl'
+      responses:
+        200:
+          description: OK
+
+    delete:
+      tags:
+        - Acls
+      summary: deleteAcl
+      operationId: deleteAcl
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/KafkaAcl'
+      responses:
+        200:
+          description: OK
+        404:
+          description: Acl not found
+
+  /api/clusters/{clusterName}/acl/consumer:
+    post:
+      tags:
+        - Acls
+      summary: createConsumerAcl
+      operationId: createConsumerAcl
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CreateConsumerAcl'
+      responses:
+        200:
+          description: OK
+
+  /api/clusters/{clusterName}/acl/producer:
+    post:
+      tags:
+        - Acls
+      summary: createProducerAcl
+      operationId: createProducerAcl
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CreateProducerAcl'
+      responses:
+        200:
+          description: OK
+
+  /api/clusters/{clusterName}/acl/streamApp:
+    post:
+      tags:
+        - Acls
+      summary: createStreamAppAcl
+      operationId: createStreamAppAcl
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CreateStreamAppAcl'
+      responses:
+        200:
+          description: OK
+
   /api/authorization:
     get:
       tags:
@@ -1959,7 +2160,7 @@ paths:
               properties:
                 file:
                   type: string
-                  format: filepart
+                  format: binary
       responses:
         200:
           description: OK
@@ -2057,6 +2258,26 @@ components:
             type: string
             enum:
               - DYNAMIC_CONFIG
+        build:
+          type: object
+          properties:
+            commitId:
+              type: string
+            version:
+              type: string
+            buildTime:
+              type: string
+            isLatestRelease:
+              type: boolean
+        latestRelease:
+          type: object
+          properties:
+            versionTag:
+              type: string
+            publishedAt:
+              type: string
+            htmlUrl:
+              type: string
 
     Cluster:
       type: object
@@ -2092,6 +2313,8 @@ components:
               - KAFKA_CONNECT
               - KSQL_DB
               - TOPIC_DELETION
+              - KAFKA_ACL_VIEW # get ACLs listing
+              - KAFKA_ACL_EDIT # create & delete ACLs
       required:
         - id
         - name
@@ -2495,6 +2718,16 @@ components:
           type: number
         bytesOutPerSec:
           type: number
+        partitionsLeader:
+          type: integer
+        partitions:
+          type: integer
+        inSyncPartitions:
+          type: integer
+        partitionsSkew:
+          type: number
+        leadersSkew:
+          type: number
       required:
         - id
 
@@ -2552,6 +2785,10 @@ components:
           format: int64
 
     ConsumerGroup:
+      discriminator:
+        propertyName: inherit
+        mapping:
+          details: "#/components/schemas/ConsumerGroupDetails"
       type: object
       properties:
         groupId:
@@ -2568,7 +2805,7 @@ components:
           $ref: "#/components/schemas/ConsumerGroupState"
         coordinator:
           $ref: "#/components/schemas/Broker"
-        messagesBehind:
+        consumerLag:
           type: integer
           format: int64
           description: null if consumer group has no offsets committed
@@ -2581,6 +2818,8 @@ components:
         - NAME
         - MEMBERS
         - STATE
+        - MESSAGES_BEHIND
+        - TOPIC_NUM
 
     ConsumerGroupsPageResponse:
       type: object
@@ -2592,6 +2831,37 @@ components:
           items:
             $ref: '#/components/schemas/ConsumerGroup'
 
+    SmartFilterTestExecution:
+      type: object
+      required: [filterCode]
+      properties:
+        filterCode:
+          type: string
+        key:
+          type: string
+        value:
+          type: string
+        headers:
+          type: object
+          additionalProperties:
+            type: string
+        partition:
+          type: integer
+        offset:
+          type: integer
+          format: int64
+        timestampMs:
+          type: integer
+          format: int64
+
+    SmartFilterTestExecutionResult:
+      type: object
+      properties:
+        result:
+          type: boolean
+        error:
+          type: string
+
     CreateTopicMessage:
       type: object
       properties:
@@ -2813,7 +3083,7 @@ components:
         endOffset:
           type: integer
           format: int64
-        messagesBehind:
+        consumerLag:
           type: integer
           format: int64
           description: null if consumer group has no offsets committed
@@ -2903,6 +3173,10 @@ components:
           type: string
         schemaType:
           $ref: '#/components/schemas/SchemaType'
+        references:
+          type: array
+          items:
+            $ref: '#/components/schemas/SchemaReference'
       required:
         - id
         - subject
@@ -2920,13 +3194,30 @@ components:
         schema:
           type: string
         schemaType:
-          $ref: '#/components/schemas/SchemaType'
-          # upon updating a schema, the type of existing schema can't be changed
+          $ref: '#/components/schemas/SchemaType' # upon updating a schema, the type of existing schema can't be changed
+        references:
+          type: array
+          items:
+            $ref: '#/components/schemas/SchemaReference'
       required:
         - subject
         - schema
         - schemaType
 
+    SchemaReference:
+      type: object
+      properties:
+        name:
+          type: string
+        subject:
+          type: string
+        version:
+          type: integer
+      required:
+        - name
+        - subject
+        - version
+
     CompatibilityLevel:
       type: object
       properties:
@@ -3489,6 +3780,7 @@ components:
         - MESSAGES_READ
         - MESSAGES_PRODUCE
         - MESSAGES_DELETE
+        - RESTART
 
     ResourceType:
       type: string
@@ -3500,6 +3792,126 @@ components:
         - SCHEMA
         - CONNECT
         - KSQL
+        - ACL
+        - AUDIT
+
+    KafkaAcl:
+      type: object
+      required: [resourceType, resourceName, namePatternType, principal, host, operation, permission]
+      properties:
+        resourceType:
+          $ref: '#/components/schemas/KafkaAclResourceType'
+        resourceName:
+          type: string # "*" if acl can be applied to any resource of given type
+        namePatternType:
+          $ref: '#/components/schemas/KafkaAclNamePatternType'
+        principal:
+          type: string
+        host:
+          type: string
+        operation:
+          type: string
+          enum:
+            - UNKNOWN # Unknown operation, need to update mapping code on BE
+            - ALL # Cluster, Topic, Group
+            - READ  # Topic, Group
+            - WRITE # Topic, TransactionalId
+            - CREATE # Cluster, Topic
+            - DELETE  # Topic, Group
+            - ALTER  # Cluster, Topic,
+            - DESCRIBE # Cluster, Topic, Group, TransactionalId, DelegationToken
+            - CLUSTER_ACTION # Cluster
+            - DESCRIBE_CONFIGS # Cluster, Topic
+            - ALTER_CONFIGS   # Cluster, Topic
+            - IDEMPOTENT_WRITE # Cluster
+            - CREATE_TOKENS
+            - DESCRIBE_TOKENS
+        permission:
+          type: string
+          enum:
+            - ALLOW
+            - DENY
+
+    CreateConsumerAcl:
+      type: object
+      required: [principal, host]
+      properties:
+        principal:
+          type: string
+        host:
+          type: string
+        topics:
+          type: array
+          items:
+            type: string
+        topicsPrefix:
+          type: string
+        consumerGroups:
+          type: array
+          items:
+            type: string
+        consumerGroupsPrefix:
+          type: string
+
+    CreateProducerAcl:
+      type: object
+      required: [principal, host]
+      properties:
+        principal:
+          type: string
+        host:
+          type: string
+        topics:
+          type: array
+          items:
+            type: string
+        topicsPrefix:
+          type: string
+        transactionalId:
+          type: string
+        transactionsIdPrefix:
+          type: string
+        idempotent:
+          type: boolean
+          default: false
+
+    CreateStreamAppAcl:
+      type: object
+      required: [principal, host, applicationId, inputTopics, outputTopics]
+      properties:
+        principal:
+          type: string
+        host:
+          type: string
+        inputTopics:
+          type: array
+          items:
+            type: string
+        outputTopics:
+          type: array
+          items:
+            type: string
+        applicationId:
+          nullable: false
+          type: string
+
+    KafkaAclResourceType:
+      type: string
+      enum:
+        - UNKNOWN # Unknown operation, need to update mapping code on BE
+        - TOPIC
+        - GROUP
+        - CLUSTER
+        - TRANSACTIONAL_ID
+        - DELEGATION_TOKEN
+        - USER
+
+    KafkaAclNamePatternType:
+      type: string
+      enum:
+        - MATCH
+        - LITERAL
+        - PREFIXED
 
     RestartRequest:
       type: object
@@ -3636,9 +4048,28 @@ components:
                               type: array
                               items:
                                 $ref: '#/components/schemas/Action'
+            webclient:
+              type: object
+              properties:
+                maxInMemoryBufferSize:
+                  type: string
+                  description: "examples: 20, 12KB, 5MB"
             kafka:
               type: object
               properties:
+                polling:
+                  type: object
+                  properties:
+                    pollTimeoutMs:
+                      type: integer
+                    maxPageSize:
+                      type: integer
+                    defaultPageSize:
+                      type: integer
+                adminClientTimeout:
+                  type: integer
+                internalTopicPrefix:
+                  type: string
                 clusters:
                   type: array
                   items:
@@ -3679,7 +4110,7 @@ components:
                           keystoreLocation:
                             type: string
                           keystorePassword:
-                              type: string
+                            type: string
                       ksqldbServerAuth:
                         type: object
                         properties:
@@ -3767,7 +4198,9 @@ components:
                               type: array
                               items:
                                 type: string
-                            pattern:
+                            fieldsNamePattern:
+                              type: string
+                            maskingCharsReplacement:
                               type: array
                               items:
                                 type: string
@@ -3780,3 +4213,21 @@ components:
                       pollingThrottleRate:
                         type: integer
                         format: int64
+                      audit:
+                        type: object
+                        properties:
+                          level:
+                            type: string
+                            enum: [ "ALL", "ALTER_ONLY" ]
+                          topic:
+                            type: string
+                          auditTopicsPartitions:
+                            type: integer
+                          topicAuditEnabled:
+                            type: boolean
+                          consoleAuditEnabled:
+                            type: boolean
+                          auditTopicProperties:
+                            type: object
+                            additionalProperties:
+                              type: string