Repository: provectus/kafka-ui
Issue: KC: NPE on sanitizing
Labels: type/bug
scope/backend
status/accepted

### Issue submitter TODO list

- [X] I've looked up my issue in [FAQ](https://docs.kafka-ui.provectus.io/faq/common-problems)
- [X] I've searched for an already existing issues [here](https://github.com/provectus/kafka-ui/issues)
- [X] I've tried running `master`-labeled docker image and the issue still persists there
- [X] I'm running a supported version of the application which is listed [here](https://github.com/provectus/kafka-ui/blob/master/SECURITY.md)

### Describe the bug (actual behavior)

Unable to load connectors page in kafka-ui
we have four clusters and this is happening only on one cluster
``` java
500 Server Error for HTTP GET "/api/clusters/dev/connectors"
java.lang.NullPointerException: null
	at java.base/java.util.Objects.requireNonNull(Objects.java:208)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
	*__checkpoint ⇢ Handler com.provectus.kafka.ui.controller.KafkaConnectController#getAllConnectors(String, String, ConnectorColumnsToSortDTO, SortOrderDTO, ServerWebExchange) [DispatcherHandler]
	*__checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ AuthorizationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ExceptionTranslationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ LogoutWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ServerRequestCacheWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ HttpHeaderWriterWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.web.filter.reactive.ServerHttpObservationFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ HTTP GET "/api/clusters/dev/connectors" [ExceptionHandlingWebHandler]
Original Stack Trace:
		at java.base/java.util.Objects.requireNonNull(Objects.java:208)
		at java.base/java.util.stream.Collectors.lambda$uniqKeysMapAccumulator$1(Collectors.java:180)
		at java.base/java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
		at java.base/java.util.HashMap$EntrySpliterator.forEachRemaining(HashMap.java:1850)
		at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
		at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
		at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
		at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
		at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
		at com.provectus.kafka.ui.service.KafkaConnectService.lambda$getConnector$25(KafkaConnectService.java:181)
		at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
		at reactor.core.publisher.SerializedSubscriber.onNext(SerializedSubscriber.java:99)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onNext(FluxRetryWhen.java:174)
		at reactor.core.publisher.MonoFlatMap$FlatMapMain.secondComplete(MonoFlatMap.java:245)
		at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:305)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:158)
		at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:299)
		at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:337)
		at reactor.core.publisher.Operators$BaseFluxToMonoOperator.completePossiblyEmpty(Operators.java:2071)
		at reactor.core.publisher.MonoCollect$CollectSubscriber.onComplete(MonoCollect.java:145)
		at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:260)
		at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
		at reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:415)
		at reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:431)
		at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:485)
		at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:712)
		at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:113)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
		at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318)
		at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
		at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1382)
		at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1245)
		at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1294)
		at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:529)
		at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:468)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
		at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:800)
		at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:499)
		at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:397)
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
		at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
		at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
		at java.base/java.lang.Thread.run(Thread.java:833)
```


I tried setting below config from [here](https://github.com/provectus/kafka-ui/issues/3243#issuecomment-1386574743) still seeing same error
```
 - name: WEBCLIENT_MAX_IN_MEMORY_BUFFER_SIZE
    value: "12500MB"
```

### Expected behavior

Connectors page should load list of connectors if not it should say clearly what the issue is

### Your installation details

1. App version (https://github.com/provectus/kafka-ui/commit/fdd9ad9)
2. No helm chart
3. App config

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-ui
  namespace: kafka
  labels:
    app: kafka-ui
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-ui
  template:        
    metadata:
      labels:
        app: kafka-ui
    spec:
      containers:
      - name: kafka-ui
        image: provectuslabs/kafka-ui:v0.7.0
        ports:
        - containerPort: 8080
        env:     
          # dev   
          - name: KAFKA_CLUSTERS_0_NAME
            value: dev
          - name: KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS
            value: <redacted>
          - name: KAFKA_CLUSTERS_0_SCHEMAREGISTRY
            value: <redacted>
          - name: KAFKA_CLUSTERS_0_KSQLDBSERVER
            value: <redacted>
          - name: KAFKA_CLUSTERS_0_READONLY
            value: "False"
          - name: KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME
            value: dev-kafka-connect
          - name: KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS
            value: <redacted>
          - name: KAFKA_CLUSTERS_0_KAFKACONNECT_1_NAME
            value: dev-sink-s3-connect
          - name: KAFKA_CLUSTERS_0_KAFKACONNECT_1_ADDRESS
            value: <redacted>
          - name: KAFKA_CLUSTERS_0_KAFKACONNECT_2_NAME
            value: dev-source-documentdb-connect
          - name: KAFKA_CLUSTERS_0_KAFKACONNECT_2_ADDRESS
            value: <redacted>

          # yyyy
          - name: KAFKA_CLUSTERS_1_NAME
            value: stage
          - name: KAFKA_CLUSTERS_1_BOOTSTRAPSERVERS
            value: <redacted>
          - name: KAFKA_CLUSTERS_1_SCHEMAREGISTRY
            value: <redacted>
          - name: KAFKA_CLUSTERS_1_KSQLDBSERVER
            value: <redacted>
          - name: KAFKA_CLUSTERS_1_READONLY
            value: "False"
          - name: KAFKA_CLUSTERS_1_KAFKACONNECT_0_NAME
            value: yyyyy-kafka-connect
          - name: KAFKA_CLUSTERS_1_KAFKACONNECT_0_ADDRESS
            value: <redacted>
          - name: KAFKA_CLUSTERS_1_KAFKACONNECT_1_NAME
            value: yyyy-sink-s3-connect
          - name: KAFKA_CLUSTERS_1_KAFKACONNECT_1_ADDRESS
            value: <redacted>
          - name: KAFKA_CLUSTERS_1_KAFKACONNECT_2_NAME
            value: yyyy-source-documentdb-connect
          - name: KAFKA_CLUSTERS_1_KAFKACONNECT_2_ADDRESS
            value: <redacted>

          # zzzzz
          - name: KAFKA_CLUSTERS_2_NAME
            value: preprod
          - name: KAFKA_CLUSTERS_2_BOOTSTRAPSERVERS
            value: <redacted>
          - name: KAFKA_CLUSTERS_2_SCHEMAREGISTRY
            value: <redacted>
          - name: KAFKA_CLUSTERS_2_KSQLDBSERVER
            value: <redacted>
          - name: KAFKA_CLUSTERS_2_READONLY
            value: "False"
          - name: KAFKA_CLUSTERS_2_KAFKACONNECT_0_NAME
            value: zzzz-kafka-connect
          - name: KAFKA_CLUSTERS_2_KAFKACONNECT_0_ADDRESS
            value: <redacted>
          - name: KAFKA_CLUSTERS_2_KAFKACONNECT_1_NAME
            value: zzzz-sink-s3-connect
          - name: KAFKA_CLUSTERS_2_KAFKACONNECT_1_ADDRESS
            value: <redacted>
          - name: KAFKA_CLUSTERS_2_KAFKACONNECT_2_NAME
            value: zzzz-source-documentdb-connect
          - name: KAFKA_CLUSTERS_2_KAFKACONNECT_2_ADDRESS
            value: <redacted>

          # xxxxx
          - name: KAFKA_CLUSTERS_3_NAME
            value: xxxxx
          - name: KAFKA_CLUSTERS_3_BOOTSTRAPSERVERS
            value: <redacted>
          - name: KAFKA_CLUSTERS_3_SCHEMAREGISTRY
            value: <redacted>
          - name: KAFKA_CLUSTERS_3_KSQLDBSERVER
            value: <redacted>
          - name: KAFKA_CLUSTERS_3_READONLY
            value: "False"
          - name: KAFKA_CLUSTERS_3_KAFKACONNECT_0_NAME
            value: xxxx-kafka-connect
          - name: KAFKA_CLUSTERS_3_KAFKACONNECT_0_ADDRESS
            value: <redacted>
          - name: KAFKA_CLUSTERS_3_KAFKACONNECT_1_NAME
            value: xxxx-sink-s3-connect
          - name: KAFKA_CLUSTERS_3_KAFKACONNECT_1_ADDRESS
            value: <redacted>
          - name: KAFKA_CLUSTERS_3_KAFKACONNECT_2_NAME
            value: xxxx-source-documentdb-connect
          - name: KAFKA_CLUSTERS_3_KAFKACONNECT_2_ADDRESS
            value: <redacted>
          - name: WEBCLIENT_MAX_IN_MEMORY_BUFFER_SIZE
            value: "250MB"
```

### Steps to reproduce

Not sure how to reproduce. This is happening only on specific cluster which has 110 connectors

### Screenshots

_No response_

### Logs

``` java
500 Server Error for HTTP GET "/api/clusters/dev/connectors"
java.lang.NullPointerException: null
	at java.base/java.util.Objects.requireNonNull(Objects.java:208)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
	*__checkpoint ⇢ Handler com.provectus.kafka.ui.controller.KafkaConnectController#getAllConnectors(String, String, ConnectorColumnsToSortDTO, SortOrderDTO, ServerWebExchange) [DispatcherHandler]
	*__checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ AuthorizationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ExceptionTranslationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ LogoutWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ServerRequestCacheWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ HttpHeaderWriterWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.web.filter.reactive.ServerHttpObservationFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ HTTP GET "/api/clusters/dev/connectors" [ExceptionHandlingWebHandler]
Original Stack Trace:
		at java.base/java.util.Objects.requireNonNull(Objects.java:208)
		at java.base/java.util.stream.Collectors.lambda$uniqKeysMapAccumulator$1(Collectors.java:180)
		at java.base/java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
		at java.base/java.util.HashMap$EntrySpliterator.forEachRemaining(HashMap.java:1850)
		at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
		at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
		at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
		at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
		at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
		at com.provectus.kafka.ui.service.KafkaConnectService.lambda$getConnector$25(KafkaConnectService.java:181)
		at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
		at reactor.core.publisher.SerializedSubscriber.onNext(SerializedSubscriber.java:99)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onNext(FluxRetryWhen.java:174)
		at reactor.core.publisher.MonoFlatMap$FlatMapMain.secondComplete(MonoFlatMap.java:245)
		at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:305)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:158)
		at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:299)
		at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:337)
		at reactor.core.publisher.Operators$BaseFluxToMonoOperator.completePossiblyEmpty(Operators.java:2071)
		at reactor.core.publisher.MonoCollect$CollectSubscriber.onComplete(MonoCollect.java:145)
		at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:260)
		at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
		at reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:415)
		at reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:431)
		at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:485)
		at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:712)
		at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:113)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
		at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318)
		at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
		at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1382)
		at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1245)
		at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1294)
		at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:529)
		at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:468)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
		at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
		at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:800)
		at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:499)
		at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:397)
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
		at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
		at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
		at java.base/java.lang.Thread.run(Thread.java:833)
```

### Additional context

_No response_

Issue comments:
Hello there sappusaketh! 👋

Thank you and congratulations 🎉 for opening your very first issue in this project! 💖

In case you want to claim this issue, please comment down below! We will try to get back to you as soon as we can. 👀

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConfigSanitizer.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConfigSanitizer.java
index 375afb0fefd..b4cdf144c9f 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConfigSanitizer.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConfigSanitizer.java
@@ -5,11 +5,13 @@
 import com.google.common.collect.ImmutableList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.config.SslConfigs;
@@ -17,7 +19,7 @@
 import org.springframework.stereotype.Component;
 
 @Component
-class KafkaConfigSanitizer  {
+class KafkaConfigSanitizer {
 
   private static final String SANITIZED_VALUE = "******";
 
@@ -65,10 +67,8 @@ private static Set<String> kafkaConfigKeysToSanitize() {
         .collect(Collectors.toSet());
   }
 
-  public Object sanitize(String key, Object value) {
-    if (value == null) {
-      return null;
-    }
+  @Nullable
+  public Object sanitize(String key, @Nullable Object value) {
     for (Pattern pattern : sanitizeKeysPatterns) {
       if (pattern.matcher(key).matches()) {
         return SANITIZED_VALUE;
@@ -77,5 +77,12 @@ public Object sanitize(String key, Object value) {
     return value;
   }
 
+  public Map<String, Object> sanitizeConnectorConfig(@Nullable Map<String, Object> original) {
+    var result = new HashMap<String, Object>(); //null-values supporting map!
+    if (original != null) {
+      original.forEach((k, v) -> result.put(k, sanitize(k, v)));
+    }
+    return result;
+  }
 
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java
index 98b61541c5f..390348707d3 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java
@@ -24,7 +24,6 @@
 import com.provectus.kafka.ui.model.TaskDTO;
 import com.provectus.kafka.ui.model.connect.InternalConnectInfo;
 import com.provectus.kafka.ui.util.ReactiveFailover;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -176,19 +175,14 @@ public Mono<ConnectorDTO> getConnector(KafkaCluster cluster, String connectName,
                         e -> emptyStatus(connectorName))
                     .map(connectorStatus -> {
                       var status = connectorStatus.getConnector();
-                      final Map<String, Object> obfuscatedConfig = connector.getConfig().entrySet()
-                          .stream()
-                          .collect(Collectors.toMap(
-                              Map.Entry::getKey,
-                              e -> kafkaConfigSanitizer.sanitize(e.getKey(), e.getValue())
-                          ));
-                      ConnectorDTO result = (ConnectorDTO) new ConnectorDTO()
+                      var sanitizedConfig = kafkaConfigSanitizer.sanitizeConnectorConfig(connector.getConfig());
+                      ConnectorDTO result = new ConnectorDTO()
                           .connect(connectName)
                           .status(kafkaConnectMapper.fromClient(status))
                           .type(connector.getType())
                           .tasks(connector.getTasks())
                           .name(connector.getName())
-                          .config(obfuscatedConfig);
+                          .config(sanitizedConfig);
 
                       if (connectorStatus.getTasks() != null) {
                         boolean isAnyTaskFailed = connectorStatus.getTasks().stream()
@@ -217,12 +211,7 @@ public Mono<Map<String, Object>> getConnectorConfig(KafkaCluster cluster, String
                                                       String connectorName) {
     return api(cluster, connectName)
         .mono(c -> c.getConnectorConfig(connectorName))
-        .map(connectorConfig -> {
-          final Map<String, Object> obfuscatedMap = new HashMap<>();
-          connectorConfig.forEach((key, value) ->
-              obfuscatedMap.put(key, kafkaConfigSanitizer.sanitize(key, value)));
-          return obfuscatedMap;
-        });
+        .map(kafkaConfigSanitizer::sanitizeConnectorConfig);
   }
 
   public Mono<ConnectorDTO> setConnectorConfig(KafkaCluster cluster, String connectName,


Test patch:
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KafkaConfigSanitizerTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KafkaConfigSanitizerTest.java
index 6454cd9f2a4..9cab6b2f13f 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KafkaConfigSanitizerTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KafkaConfigSanitizerTest.java
@@ -3,14 +3,16 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Arrays;
-import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import org.junit.jupiter.api.Test;
 
 class KafkaConfigSanitizerTest {
 
   @Test
   void doNothingIfEnabledPropertySetToFalse() {
-    final var sanitizer = new KafkaConfigSanitizer(false, Collections.emptyList());
+    final var sanitizer = new KafkaConfigSanitizer(false, List.of());
     assertThat(sanitizer.sanitize("password", "secret")).isEqualTo("secret");
     assertThat(sanitizer.sanitize("sasl.jaas.config", "secret")).isEqualTo("secret");
     assertThat(sanitizer.sanitize("database.password", "secret")).isEqualTo("secret");
@@ -18,7 +20,7 @@ void doNothingIfEnabledPropertySetToFalse() {
 
   @Test
   void obfuscateCredentials() {
-    final var sanitizer = new KafkaConfigSanitizer(true, Collections.emptyList());
+    final var sanitizer = new KafkaConfigSanitizer(true, List.of());
     assertThat(sanitizer.sanitize("sasl.jaas.config", "secret")).isEqualTo("******");
     assertThat(sanitizer.sanitize("consumer.sasl.jaas.config", "secret")).isEqualTo("******");
     assertThat(sanitizer.sanitize("producer.sasl.jaas.config", "secret")).isEqualTo("******");
@@ -36,7 +38,7 @@ void obfuscateCredentials() {
 
   @Test
   void notObfuscateNormalConfigs() {
-    final var sanitizer = new KafkaConfigSanitizer(true, Collections.emptyList());
+    final var sanitizer = new KafkaConfigSanitizer(true, List.of());
     assertThat(sanitizer.sanitize("security.protocol", "SASL_SSL")).isEqualTo("SASL_SSL");
     final String[] bootstrapServer = new String[] {"test1:9092", "test2:9092"};
     assertThat(sanitizer.sanitize("bootstrap.servers", bootstrapServer)).isEqualTo(bootstrapServer);
@@ -52,4 +54,22 @@ void obfuscateCredentialsWithDefinedPatterns() {
     assertThat(sanitizer.sanitize("database.password", "no longer credential"))
             .isEqualTo("no longer credential");
   }
+
+  @Test
+  void sanitizeConnectorConfigDoNotFailOnNullableValues() {
+    Map<String, Object> originalConfig = new HashMap<>();
+    originalConfig.put("password", "secret");
+    originalConfig.put("asIs", "normal");
+    originalConfig.put("nullVal", null);
+
+    var sanitizedConfig = new KafkaConfigSanitizer(true, List.of())
+        .sanitizeConnectorConfig(originalConfig);
+
+    assertThat(sanitizedConfig)
+        .hasSize(3)
+        .containsEntry("password", "******")
+        .containsEntry("asIs", "normal")
+        .containsEntry("nullVal", null);
+  }
+
 }