Limit kafka memory usage (#8018)

Author: laurit

instrumentation/kafka/kafka-clients/kafka-clients-0.11/testing/src/main/java/io/opentelemetry/instrumentation/kafka/internal/AbstractOpenTelemetryMetricsReporterTest.java

@@ -76,6 +76,7 @@ void beforeAll() {
 
     kafka =
         new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"))
+            .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m -Xmx256m")
             .withLogConsumer(new Slf4jLogConsumer(logger))
             .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
             .withStartupTimeout(Duration.ofMinutes(1));

instrumentation/kafka/kafka-clients/kafka-clients-0.11/testing/src/main/java/io/opentelemetry/instrumentation/kafka/internal/KafkaClientBaseTest.java

@@ -65,6 +65,7 @@ public abstract class KafkaClientBaseTest {
   void setupClass() throws ExecutionException, InterruptedException, TimeoutException {
     kafka =
         new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"))
+            .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m -Xmx256m")
             .withLogConsumer(new Slf4jLogConsumer(logger))
             .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
             .withStartupTimeout(Duration.ofMinutes(1));

instrumentation/kafka/kafka-streams-0.11/javaagent/src/test/groovy/KafkaStreamsBaseTest.groovy

@@ -4,7 +4,6 @@
  */
 
 import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
-import java.time.Duration
 import org.apache.kafka.clients.admin.AdminClient
 import org.apache.kafka.clients.admin.NewTopic
 import org.apache.kafka.clients.consumer.Consumer
@@ -22,8 +21,10 @@ import org.slf4j.LoggerFactory
 import org.testcontainers.containers.KafkaContainer
 import org.testcontainers.containers.output.Slf4jLogConsumer
 import org.testcontainers.containers.wait.strategy.Wait
+import org.testcontainers.utility.DockerImageName
 import spock.lang.Shared
 
+import java.time.Duration
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 
@@ -43,7 +44,8 @@ class KafkaStreamsBaseTest extends AgentInstrumentationSpecification {
   static CountDownLatch consumerReady = new CountDownLatch(1)
 
   def setupSpec() {
-    kafka = new KafkaContainer()
+    kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"))
+      .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m -Xmx256m")
       .withLogConsumer(new Slf4jLogConsumer(logger))
       .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
       .withStartupTimeout(Duration.ofMinutes(1))

instrumentation/spring/spring-boot-autoconfigure/src/test/java/io/opentelemetry/instrumentation/spring/autoconfigure/kafka/KafkaIntegrationTest.java

@@ -47,6 +47,7 @@ class KafkaIntegrationTest {
   static void setUpKafka() {
     kafka =
         new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.1.9"))
+            .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m -Xmx256m")
             .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
             .withStartupTimeout(Duration.ofMinutes(1));
     kafka.start();

instrumentation/spring/spring-kafka-2.7/testing/src/main/java/io/opentelemetry/testing/AbstractSpringKafkaTest.java

@@ -47,6 +47,7 @@ public abstract class AbstractSpringKafkaTest {
   static void setUpKafka() {
     kafka =
         new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"))
+            .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m -Xmx256m")
             .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
             .withStartupTimeout(Duration.ofMinutes(1));
     kafka.start();

instrumentation/vertx/vertx-kafka-client-3.6/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/vertx/kafka/v3_6/AbstractVertxKafkaTest.java

@@ -61,6 +61,7 @@ public abstract class AbstractVertxKafkaTest {
   static void setUpAll() {
     kafka =
         new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"))
+            .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m -Xmx256m")
             .withLogConsumer(new Slf4jLogConsumer(logger))
             .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
             .withStartupTimeout(Duration.ofMinutes(1));