Merge branch 'main' into issues/53

Author: Haarolean

api/src/main/java/io/kafbat/ui/controller/ApplicationConfigController.java

@@ -5,6 +5,7 @@
 
 import io.kafbat.ui.api.ApplicationConfigApi;
 import io.kafbat.ui.config.ClustersProperties;
+import io.kafbat.ui.model.ActionDTO;
 import io.kafbat.ui.model.ApplicationConfigDTO;
 import io.kafbat.ui.model.ApplicationConfigPropertiesDTO;
 import io.kafbat.ui.model.ApplicationConfigValidationDTO;
@@ -18,6 +19,7 @@
 import io.kafbat.ui.util.ApplicationRestarter;
 import io.kafbat.ui.util.DynamicConfigOperations;
 import java.util.Map;
+import java.util.Optional;
 import javax.annotation.Nullable;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -46,6 +48,12 @@ interface PropertiesMapper {
     DynamicConfigOperations.PropertiesStructure fromDto(ApplicationConfigPropertiesDTO dto);
 
     ApplicationConfigPropertiesDTO toDto(DynamicConfigOperations.PropertiesStructure propertiesStructure);
+
+    default ActionDTO stringToActionDto(String str) {
+      return Optional.ofNullable(str)
+          .map(s -> Enum.valueOf(ActionDTO.class, s.toUpperCase()))
+          .orElseThrow();
+    }
   }
 
   private final DynamicConfigOperations dynamicConfigOperations;
@@ -75,7 +83,7 @@ public Mono<ResponseEntity<ApplicationConfigDTO>> getCurrentConfig(ServerWebExch
   @Override
   public Mono<ResponseEntity<Void>> restartWithConfig(Mono<RestartRequestDTO> restartRequestDto,
                                                       ServerWebExchange exchange) {
-    var context =  AccessContext.builder()
+    var context = AccessContext.builder()
         .applicationConfigActions(EDIT)
         .operationName("restartWithConfig")
         .build();

api/src/main/java/io/kafbat/ui/emitter/OffsetsInfo.java

@@ -6,11 +6,13 @@
 import java.util.HashSet;
 import java.util.Map;
 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;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 
 @Slf4j
 @Getter
@@ -34,7 +36,7 @@ class OffsetsInfo {
 
   OffsetsInfo(Consumer<?, ?> consumer, Collection<TopicPartition> targetPartitions) {
     this.consumer = consumer;
-    this.beginOffsets = consumer.beginningOffsets(targetPartitions);
+    this.beginOffsets = firstOffsetsForPolling(consumer, targetPartitions);
     this.endOffsets = consumer.endOffsets(targetPartitions);
     endOffsets.forEach((tp, endOffset) -> {
       var beginningOffset = beginOffsets.get(tp);
@@ -46,6 +48,28 @@ class OffsetsInfo {
     });
   }
 
+
+  private Map<TopicPartition, Long> firstOffsetsForPolling(Consumer<?, ?> consumer,
+                                                           Collection<TopicPartition> partitions) {
+    try {
+      // we try to use offsetsForTimes() to find earliest offsets, since for
+      // some topics (like compacted) beginningOffsets() ruturning 0 offsets
+      // even when effectively first offset can be very high
+      var offsets = consumer.offsetsForTimes(
+          partitions.stream().collect(Collectors.toMap(p -> p, p -> 0L))
+      );
+      // result of offsetsForTimes() can be null, if message version < 0.10.0
+      if (offsets.entrySet().stream().noneMatch(e -> e.getValue() == null)) {
+        return offsets.entrySet().stream()
+            .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));
+      }
+    } catch (UnsupportedOperationException | UnsupportedVersionException e) {
+      // offsetsForTimes() not supported
+    }
+    //falling back to beginningOffsets() if offsetsForTimes() not supported
+    return consumer.beginningOffsets(partitions);
+  }
+
   boolean assignedPartitionsFullyPolled() {
     for (var tp : consumer.assignment()) {
       Preconditions.checkArgument(endOffsets.containsKey(tp));

api/src/main/java/io/kafbat/ui/emitter/RangePollingEmitter.java

@@ -3,7 +3,9 @@
 import io.kafbat.ui.model.ConsumerPosition;
 import io.kafbat.ui.model.TopicMessageEventDTO;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
@@ -84,20 +86,22 @@ private List<ConsumerRecord<Bytes, Bytes>> poll(EnhancedConsumer consumer,
     range.forEach((tp, fromTo) -> consumer.seek(tp, fromTo.from));
 
     List<ConsumerRecord<Bytes, Bytes>> result = new ArrayList<>();
-    while (!sink.isCancelled() && consumer.paused().size() < range.size()) {
+    Set<TopicPartition> paused = new HashSet<>();
+    while (!sink.isCancelled() && paused.size() < range.size()) {
       var polledRecords = poll(sink, consumer);
       range.forEach((tp, fromTo) -> {
         polledRecords.records(tp).stream()
             .filter(r -> r.offset() < fromTo.to)
             .forEach(result::add);
 
         //next position is out of target range -> pausing partition
-        if (consumer.position(tp) >= fromTo.to) {
+        if (!paused.contains(tp) && consumer.position(tp) >= fromTo.to) {
+          paused.add(tp);
           consumer.pause(List.of(tp));
         }
       });
     }
-    consumer.resume(consumer.paused());
+    consumer.resume(paused);
     return result;
   }
 }

api/src/main/java/io/kafbat/ui/model/rbac/Permission.java

@@ -3,6 +3,7 @@
 import static org.apache.commons.collections.CollectionUtils.isNotEmpty;
 
 import com.google.common.base.Preconditions;
+import io.kafbat.ui.model.ActionDTO;
 import io.kafbat.ui.model.rbac.permission.PermissibleAction;
 import java.util.List;
 import java.util.regex.Pattern;
@@ -50,7 +51,7 @@ public void transform() {
     if (value != null) {
       this.compiledValuePattern = Pattern.compile(value);
     }
-    if (actions.stream().anyMatch("ALL"::equalsIgnoreCase)) {
+    if (actions.stream().anyMatch(ActionDTO.ALL.name()::equalsIgnoreCase)) {
       this.parsedActions = resource.allActions();
     } else {
       this.parsedActions = resource.parseActionsWithDependantsUnnest(actions);

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

@@ -3821,6 +3821,7 @@ components:
     Action:
       type: string
       enum:
+        - ALL
         - VIEW
         - EDIT
         - CREATE

frontend/src/components/Connect/Details/Actions/Actions.tsx

@@ -78,7 +78,7 @@ const Actions: React.FC = () => {
             permission={{
               resource: ResourceType.CONNECT,
               action: Action.EDIT,
-              value: routerProps.connectorName,
+              value: routerProps.connectName,
             }}
           >
             Pause
@@ -91,7 +91,7 @@ const Actions: React.FC = () => {
             permission={{
               resource: ResourceType.CONNECT,
               action: Action.EDIT,
-              value: routerProps.connectorName,
+              value: routerProps.connectName,
             }}
           >
             Resume
@@ -103,7 +103,7 @@ const Actions: React.FC = () => {
           permission={{
             resource: ResourceType.CONNECT,
             action: Action.RESTART,
-            value: routerProps.connectorName,
+            value: routerProps.connectName,
           }}
         >
           Restart Connector
@@ -114,7 +114,7 @@ const Actions: React.FC = () => {
           permission={{
             resource: ResourceType.CONNECT,
             action: Action.RESTART,
-            value: routerProps.connectorName,
+            value: routerProps.connectName,
           }}
         >
           Restart All Tasks
@@ -125,7 +125,7 @@ const Actions: React.FC = () => {
           permission={{
             resource: ResourceType.CONNECT,
             action: Action.RESTART,
-            value: routerProps.connectorName,
+            value: routerProps.connectName,
           }}
         >
           Restart Failed Tasks
@@ -139,7 +139,7 @@ const Actions: React.FC = () => {
           permission={{
             resource: ResourceType.CONNECT,
             action: Action.DELETE,
-            value: routerProps.connectorName,
+            value: routerProps.connectName,
           }}
         >
           Delete

frontend/src/components/Connect/Details/Tasks/ActionsCellTasks.tsx

@@ -26,7 +26,7 @@ const ActionsCellTasks: React.FC<CellContext<Task, unknown>> = ({ row }) => {
         permission={{
           resource: ResourceType.CONNECT,
           action: Action.RESTART,
-          value: routerProps.connectorName,
+          value: routerProps.connectName,
         }}
       >
         <span>Restart task</span>

frontend/src/components/Connect/List/ActionsCell.tsx

@@ -67,7 +67,7 @@ const ActionsCell: React.FC<CellContext<FullConnectorInfo, unknown>> = ({
           permission={{
             resource: ResourceType.CONNECT,
             action: Action.EDIT,
-            value: name,
+            value: connect,
           }}
         >
           Resume
@@ -79,7 +79,7 @@ const ActionsCell: React.FC<CellContext<FullConnectorInfo, unknown>> = ({
         permission={{
           resource: ResourceType.CONNECT,
           action: Action.RESTART,
-          value: name,
+          value: connect,
         }}
       >
         Restart Connector
@@ -90,7 +90,7 @@ const ActionsCell: React.FC<CellContext<FullConnectorInfo, unknown>> = ({
         permission={{
           resource: ResourceType.CONNECT,
           action: Action.RESTART,
-          value: name,
+          value: connect,
         }}
       >
         Restart All Tasks
@@ -101,7 +101,7 @@ const ActionsCell: React.FC<CellContext<FullConnectorInfo, unknown>> = ({
         permission={{
           resource: ResourceType.CONNECT,
           action: Action.RESTART,
-          value: name,
+          value: connect,
         }}
       >
         Restart Failed Tasks