Repository: provectus/kafka-ui
Issue: Failures when Kafka Connectors has high number of tasks (limit on max bytes to buffer : 262144)
Labels: type/bug
good first issue
scope/backend
status/accepted
status/confirmed

**Describe the bug**
Stack trace:
		at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:229)
		at org.springframework.web.reactive.function.client.DefaultClientResponse.lambda$createException$2(DefaultClientResponse.java:213)
		at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
		at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2398)
		at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:2194)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onSubscribe(FluxOnErrorResume.java:74)
		at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:55)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4399)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103)
		at reactor.core.publisher.Operators$MonoSubscriber.onError(Operators.java:1863)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onError(FluxMapFuseable.java:140)
		at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onError(FluxContextWrite.java:121)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onError(FluxMapFuseable.java:334)
		at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onError(FluxFilterFuseable.java:382)
		at reactor.core.publisher.MonoCollect$CollectSubscriber.onError(MonoCollect.java:144)
		at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:132)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
		at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:132)
		at reactor.core.publisher.Operators.error(Operators.java:198)
		at reactor.netty.channel.FluxReceive.startReceiver(FluxReceive.java:182)
		at reactor.netty.channel.FluxReceive.subscribe(FluxReceive.java:143)
		at reactor.core.publisher.InternalFluxOperator.subscribe(InternalFluxOperator.java:62)
		at reactor.netty.ByteBufFlux.subscribe(ByteBufFlux.java:339)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4399)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103)
		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onError(MonoFlatMap.java:172)
		at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onError(FluxContextWrite.java:121)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onError(FluxMapFuseable.java:334)
		at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onError(FluxFilterFuseable.java:382)
		at reactor.core.publisher.MonoCollect$CollectSubscriber.onError(MonoCollect.java:144)
		at reactor.core.publisher.MonoCollect$CollectSubscriber.onNext(MonoCollect.java:123)
		at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200)
		at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
		at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:364)
		at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:404)
		at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:724)
		at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
		at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:311)
		at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:432)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
		at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
		at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
		at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:795)
		at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:480)
		at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378)
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
		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:830)
Caused by: org.springframework.core.io.buffer.DataBufferLimitException: Exceeded limit on max bytes to buffer : 262144
	at org.springframework.core.io.buffer.LimitedDataBufferList.raiseLimitException(LimitedDataBufferList.java:99)


**Set up**
Docker


**Steps to Reproduce**
Steps to reproduce the behavior:

1. Create ~50-80 elasticsearch tasks (or others)
2. Get some mapping error from elastic

**Expected behavior**
Should work

**Additional context**
Will be good to use variable to change **max-in-memory-size**.


Issue comments:
Hello there enshinov! 👋

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. 👀
Hey, thanks for reaching out. We'll make it configurable

Pull request review comments:
I'd prefer fully qualified import for lombok's annotation over spring's
I think we should refactor this code in general: lets
1. make KafkaConnectClients normal object (spring component), CACHE field will become non static. (I would personally rename class to KafkaConnectClientsFactory)
2. inject maxBuffSize value to it, not to KafkaConnectService

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KafkaConnectClients.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KafkaConnectClients.java
deleted file mode 100644
index de0c9054ae2..00000000000
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KafkaConnectClients.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package com.provectus.kafka.ui.client;
-
-import com.provectus.kafka.ui.connect.api.KafkaConnectClientApi;
-import com.provectus.kafka.ui.model.KafkaConnectCluster;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-public final class KafkaConnectClients {
-
-  private KafkaConnectClients() {
-
-  }
-
-  private static final Map<String, KafkaConnectClientApi> CACHE = new ConcurrentHashMap<>();
-
-  public static KafkaConnectClientApi withKafkaConnectConfig(KafkaConnectCluster config) {
-    return CACHE.computeIfAbsent(config.getAddress(), s -> new RetryingKafkaConnectClient(config));
-  }
-}
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KafkaConnectClientsFactory.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KafkaConnectClientsFactory.java
new file mode 100644
index 00000000000..3a5195f5cf0
--- /dev/null
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KafkaConnectClientsFactory.java
@@ -0,0 +1,22 @@
+package com.provectus.kafka.ui.client;
+
+import com.provectus.kafka.ui.connect.api.KafkaConnectClientApi;
+import com.provectus.kafka.ui.model.KafkaConnectCluster;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import org.springframework.util.unit.DataSize;
+
+@Service
+public class KafkaConnectClientsFactory {
+
+  @Value("${webclient.max-in-memory-buffer-size:20MB}")
+  private DataSize maxBuffSize;
+
+  private final Map<String, KafkaConnectClientApi> cache = new ConcurrentHashMap<>();
+
+  public KafkaConnectClientApi withKafkaConnectConfig(KafkaConnectCluster config) {
+    return cache.computeIfAbsent(config.getAddress(), s -> new RetryingKafkaConnectClient(config, maxBuffSize));
+  }
+}
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/RetryingKafkaConnectClient.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/RetryingKafkaConnectClient.java
index 70716613730..b1115d0eef1 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/RetryingKafkaConnectClient.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/RetryingKafkaConnectClient.java
@@ -1,22 +1,34 @@
 package com.provectus.kafka.ui.client;
 
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import com.provectus.kafka.ui.connect.ApiClient;
+import com.provectus.kafka.ui.connect.RFC3339DateFormat;
 import com.provectus.kafka.ui.connect.api.KafkaConnectClientApi;
 import com.provectus.kafka.ui.connect.model.Connector;
 import com.provectus.kafka.ui.connect.model.NewConnector;
 import com.provectus.kafka.ui.exception.KafkaConnectConflictReponseException;
 import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.model.KafkaConnectCluster;
+import java.text.DateFormat;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.TimeZone;
 import lombok.extern.slf4j.Slf4j;
+import org.openapitools.jackson.nullable.JsonNullableModule;
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
+import org.springframework.http.codec.json.Jackson2JsonDecoder;
+import org.springframework.http.codec.json.Jackson2JsonEncoder;
 import org.springframework.util.MultiValueMap;
+import org.springframework.util.unit.DataSize;
 import org.springframework.web.client.RestClientException;
+import org.springframework.web.reactive.function.client.ExchangeStrategies;
+import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.reactive.function.client.WebClientResponseException;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -27,8 +39,8 @@ public class RetryingKafkaConnectClient extends KafkaConnectClientApi {
   private static final int MAX_RETRIES = 5;
   private static final Duration RETRIES_DELAY = Duration.ofMillis(200);
 
-  public RetryingKafkaConnectClient(KafkaConnectCluster config) {
-    super(new RetryingApiClient(config));
+  public RetryingKafkaConnectClient(KafkaConnectCluster config, DataSize maxBuffSize) {
+    super(new RetryingApiClient(config, maxBuffSize));
   }
 
   private static Retry conflictCodeRetry() {
@@ -73,13 +85,48 @@ public Mono<Connector> setConnectorConfig(String connectorName, Map<String, Obje
 
   private static class RetryingApiClient extends ApiClient {
 
-    public RetryingApiClient(KafkaConnectCluster config) {
-      super();
+    private static final DateFormat dateFormat = getDefaultDateFormat();
+    private static final ObjectMapper mapper = buildObjectMapper(dateFormat);
+
+    public RetryingApiClient(KafkaConnectCluster config, DataSize maxBuffSize) {
+      super(buildWebClient(mapper, maxBuffSize), mapper, dateFormat);
       setBasePath(config.getAddress());
       setUsername(config.getUserName());
       setPassword(config.getPassword());
     }
 
+    public static DateFormat getDefaultDateFormat() {
+      DateFormat dateFormat = new RFC3339DateFormat();
+      dateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
+      return dateFormat;
+    }
+
+    public static WebClient buildWebClient(ObjectMapper mapper, DataSize maxBuffSize) {
+      ExchangeStrategies strategies = ExchangeStrategies
+              .builder()
+              .codecs(clientDefaultCodecsConfigurer -> {
+                clientDefaultCodecsConfigurer.defaultCodecs()
+                        .jackson2JsonEncoder(new Jackson2JsonEncoder(mapper, MediaType.APPLICATION_JSON));
+                clientDefaultCodecsConfigurer.defaultCodecs()
+                        .jackson2JsonDecoder(new Jackson2JsonDecoder(mapper, MediaType.APPLICATION_JSON));
+                clientDefaultCodecsConfigurer.defaultCodecs()
+                        .maxInMemorySize((int) maxBuffSize.toBytes());
+              })
+              .build();
+      WebClient.Builder webClient = WebClient.builder().exchangeStrategies(strategies);
+      return webClient.build();
+    }
+
+    public static ObjectMapper buildObjectMapper(DateFormat dateFormat) {
+      ObjectMapper mapper = new ObjectMapper();
+      mapper.setDateFormat(dateFormat);
+      mapper.registerModule(new JavaTimeModule());
+      mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+      JsonNullableModule jnm = new JsonNullableModule();
+      mapper.registerModule(jnm);
+      return mapper;
+    }
+
     @Override
     public <T> Mono<T> invokeAPI(String path, HttpMethod method, Map<String, Object> pathParams,
                                  MultiValueMap<String, String> queryParams, Object body,
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 21680d3dd91..69fd7a26276 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
@@ -2,7 +2,7 @@
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.provectus.kafka.ui.client.KafkaConnectClients;
+import com.provectus.kafka.ui.client.KafkaConnectClientsFactory;
 import com.provectus.kafka.ui.connect.api.KafkaConnectClientApi;
 import com.provectus.kafka.ui.connect.model.ConnectorStatus;
 import com.provectus.kafka.ui.connect.model.ConnectorStatusConnector;
@@ -21,7 +21,6 @@
 import com.provectus.kafka.ui.model.ConnectorTaskStatusDTO;
 import com.provectus.kafka.ui.model.FullConnectorInfoDTO;
 import com.provectus.kafka.ui.model.KafkaCluster;
-import com.provectus.kafka.ui.model.KafkaConnectCluster;
 import com.provectus.kafka.ui.model.NewConnectorDTO;
 import com.provectus.kafka.ui.model.TaskDTO;
 import com.provectus.kafka.ui.model.connect.InternalConnectInfo;
@@ -51,6 +50,7 @@ public class KafkaConnectService {
   private final KafkaConnectMapper kafkaConnectMapper;
   private final ObjectMapper objectMapper;
   private final KafkaConfigSanitizer kafkaConfigSanitizer;
+  private final KafkaConnectClientsFactory kafkaConnectClientsFactory;
 
   public Mono<Flux<ConnectDTO>> getConnects(KafkaCluster cluster) {
     return Mono.just(
@@ -328,6 +328,6 @@ private Mono<KafkaConnectClientApi> withConnectClient(KafkaCluster cluster, Stri
             .filter(connect -> connect.getName().equals(connectName))
             .findFirst())
         .switchIfEmpty(Mono.error(ConnectNotFoundException::new))
-        .map(KafkaConnectClients::withKafkaConnectConfig);
+        .map(kafkaConnectClientsFactory::withKafkaConnectConfig);
   }
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java
index 9341dd30d3b..161c284c972 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java
@@ -21,6 +21,7 @@
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.json.Jackson2JsonDecoder;
 import org.springframework.util.MimeTypeUtils;
+import org.springframework.util.unit.DataSize;
 import org.springframework.web.reactive.function.client.ExchangeStrategies;
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.reactive.function.client.WebClientResponseException;
@@ -57,9 +58,11 @@ private static class KsqlRequest {
   //--------------------------------------------------------------------------------------------
 
   private final KafkaCluster cluster;
+  private final DataSize maxBuffSize;
 
-  public KsqlApiClient(KafkaCluster cluster) {
+  public KsqlApiClient(KafkaCluster cluster, DataSize maxBuffSize) {
     this.cluster = cluster;
+    this.maxBuffSize = maxBuffSize;
   }
 
   private WebClient webClient() {
@@ -75,6 +78,7 @@ private WebClient webClient() {
         })
         .build();
     return WebClient.builder()
+        .codecs(c -> c.defaultCodecs().maxInMemorySize((int) maxBuffSize.toBytes()))
         .exchangeStrategies(exchangeStrategies)
         .build();
   }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2.java
index 12a2e7d0b24..63fcede3611 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2.java
@@ -13,16 +13,23 @@
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
-import lombok.Value;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
+import org.springframework.util.unit.DataSize;
 import reactor.core.publisher.Flux;
 
 @Slf4j
 @Service
 public class KsqlServiceV2 {
 
-  @Value
+  private final DataSize maxBuffSize;
+
+  public KsqlServiceV2(@Value("${webclient.max-in-memory-buffer-size:20MB}") DataSize maxBuffSize) {
+    this.maxBuffSize = maxBuffSize;
+  }
+
+  @lombok.Value
   private static class KsqlExecuteCommand {
     KafkaCluster cluster;
     String ksql;
@@ -48,12 +55,12 @@ public Flux<KsqlResponseTable> execute(String commandId) {
       throw new ValidationException("No command registered with id " + commandId);
     }
     registeredCommands.invalidate(commandId);
-    return new KsqlApiClient(cmd.cluster)
+    return new KsqlApiClient(cmd.cluster, maxBuffSize)
         .execute(cmd.ksql, cmd.streamProperties);
   }
 
   public Flux<KsqlTableDescriptionDTO> listTables(KafkaCluster cluster) {
-    return new KsqlApiClient(cluster)
+    return new KsqlApiClient(cluster, maxBuffSize)
         .execute("LIST TABLES;", Map.of())
         .flatMap(resp -> {
           if (!resp.getHeader().equals("Tables")) {
@@ -75,7 +82,7 @@ public Flux<KsqlTableDescriptionDTO> listTables(KafkaCluster cluster) {
   }
 
   public Flux<KsqlStreamDescriptionDTO> listStreams(KafkaCluster cluster) {
-    return new KsqlApiClient(cluster)
+    return new KsqlApiClient(cluster, maxBuffSize)
         .execute("LIST STREAMS;", Map.of())
         .flatMap(resp -> {
           if (!resp.getHeader().equals("Streams")) {


Test patch:
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlApiClientTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlApiClientTest.java
index d09e32f948c..5956d730829 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlApiClientTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlApiClientTest.java
@@ -16,6 +16,7 @@
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.springframework.util.unit.DataSize;
 import org.testcontainers.shaded.org.awaitility.Awaitility;
 import org.testcontainers.utility.DockerImageName;
 import reactor.test.StepVerifier;
@@ -26,6 +27,8 @@ class KsqlApiClientTest extends AbstractIntegrationTest {
       DockerImageName.parse("confluentinc/ksqldb-server").withTag("0.24.0"))
       .withKafka(kafka);
 
+  private static final DataSize maxBuffSize = DataSize.ofMegabytes(20);
+
   @BeforeAll
   static void startContainer() {
     KSQL_DB.start();
@@ -39,7 +42,7 @@ static void stopContainer() {
   // Tutorial is here: https://ksqldb.io/quickstart.html
   @Test
   void ksqTutorialQueriesWork() {
-    var client = new KsqlApiClient(KafkaCluster.builder().ksqldbServer(KSQL_DB.url()).build());
+    var client = new KsqlApiClient(KafkaCluster.builder().ksqldbServer(KSQL_DB.url()).build(), maxBuffSize);
     execCommandSync(client,
         "CREATE STREAM riderLocations (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE) "
             + "WITH (kafka_topic='locations', value_format='json', partitions=1);",
@@ -126,4 +129,4 @@ private void execCommandSync(KsqlApiClient client, String... ksqls) {
   }
 
 
-}
\ No newline at end of file
+}
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2Test.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2Test.java
index 22c87c9aecc..d670680ea21 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2Test.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2Test.java
@@ -13,6 +13,7 @@
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.springframework.util.unit.DataSize;
 import org.testcontainers.utility.DockerImageName;
 
 class KsqlServiceV2Test extends AbstractIntegrationTest {
@@ -24,6 +25,8 @@ class KsqlServiceV2Test extends AbstractIntegrationTest {
   private static final Set<String> STREAMS_TO_DELETE = new CopyOnWriteArraySet<>();
   private static final Set<String> TABLES_TO_DELETE = new CopyOnWriteArraySet<>();
 
+  private static final DataSize maxBuffSize = DataSize.ofMegabytes(20);
+
   @BeforeAll
   static void init() {
     KSQL_DB.start();
@@ -31,7 +34,7 @@ static void init() {
 
   @AfterAll
   static void cleanup() {
-    var client = new KsqlApiClient(KafkaCluster.builder().ksqldbServer(KSQL_DB.url()).build());
+    var client = new KsqlApiClient(KafkaCluster.builder().ksqldbServer(KSQL_DB.url()).build(), maxBuffSize);
 
     TABLES_TO_DELETE.forEach(t ->
         client.execute(String.format("DROP TABLE IF EXISTS %s DELETE TOPIC;", t), Map.of())
@@ -44,7 +47,7 @@ static void cleanup() {
     KSQL_DB.stop();
   }
 
-  private final KsqlServiceV2 ksqlService = new KsqlServiceV2();
+  private final KsqlServiceV2 ksqlService = new KsqlServiceV2(maxBuffSize);
 
   @Test
   void listStreamsReturnsAllKsqlStreams() {
@@ -52,7 +55,7 @@ void listStreamsReturnsAllKsqlStreams() {
     var streamName = "stream_" + System.currentTimeMillis();
     STREAMS_TO_DELETE.add(streamName);
 
-    new KsqlApiClient(cluster)
+    new KsqlApiClient(cluster, maxBuffSize)
         .execute(
             String.format("CREATE STREAM %s ( "
                 + "  c1 BIGINT KEY, "
@@ -81,7 +84,7 @@ void listTablesReturnsAllKsqlTables() {
     var tableName = "table_" + System.currentTimeMillis();
     TABLES_TO_DELETE.add(tableName);
 
-    new KsqlApiClient(cluster)
+    new KsqlApiClient(cluster, maxBuffSize)
         .execute(
             String.format("CREATE TABLE %s ( "
                 + "   c1 BIGINT PRIMARY KEY, "
@@ -105,4 +108,4 @@ void listTablesReturnsAllKsqlTables() {
     );
   }
 
-}
\ No newline at end of file
+}