content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.clients.producer;
  
  import org.apache.kafka.clients.ApiVersions;
  import org.apache.kafka.clients.CommonClientConfigs;
  import org.apache.kafka.clients.KafkaClient;
  import org.apache.kafka.clients.LeastLoadedNode;
  import org.apache.kafka.clients.MockClient;
  import org.apache.kafka.clients.NodeApiVersions;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
  import org.apache.kafka.clients.consumer.OffsetAndMetadata;
  import org.apache.kafka.clients.producer.internals.FutureRecordMetadata;
  import org.apache.kafka.clients.producer.internals.ProduceRequestResult;
  import org.apache.kafka.clients.producer.internals.ProducerInterceptors;
  import org.apache.kafka.clients.producer.internals.ProducerMetadata;
  import org.apache.kafka.clients.producer.internals.RecordAccumulator;
  import org.apache.kafka.clients.producer.internals.Sender;
  import org.apache.kafka.clients.producer.internals.TransactionManager;
  import org.apache.kafka.clients.producer.internals.TransactionalRequestResult;
  import org.apache.kafka.common.Cluster;
  import org.apache.kafka.common.KafkaException;
  import org.apache.kafka.common.Metric;
  import org.apache.kafka.common.MetricName;
  import org.apache.kafka.common.Node;
  import org.apache.kafka.common.PartitionInfo;
  import org.apache.kafka.common.TopicIdPartition;
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.Uuid;
  import org.apache.kafka.common.config.ConfigException;
  import org.apache.kafka.common.config.SslConfigs;
  import org.apache.kafka.common.errors.ClusterAuthorizationException;
  import org.apache.kafka.common.errors.InterruptException;
  import org.apache.kafka.common.errors.InvalidTopicException;
  import org.apache.kafka.common.errors.InvalidTxnStateException;
  import org.apache.kafka.common.errors.RecordTooLargeException;
  import org.apache.kafka.common.errors.TimeoutException;
  import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
  import org.apache.kafka.common.header.Headers;
  import org.apache.kafka.common.header.internals.RecordHeader;
  import org.apache.kafka.common.header.internals.RecordHeaders;
  import org.apache.kafka.common.internals.ClusterResourceListeners;
  import org.apache.kafka.common.message.AddOffsetsToTxnResponseData;
  import org.apache.kafka.common.message.ApiVersionsResponseData;
  import org.apache.kafka.common.message.EndTxnResponseData;
  import org.apache.kafka.common.message.InitProducerIdResponseData;
  import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
  import org.apache.kafka.common.metrics.JmxReporter;
  import org.apache.kafka.common.metrics.KafkaMetric;
  import org.apache.kafka.common.metrics.Measurable;
  import org.apache.kafka.common.metrics.MetricConfig;
  import org.apache.kafka.common.metrics.Metrics;
  import org.apache.kafka.common.metrics.Monitorable;
  import org.apache.kafka.common.metrics.PluginMetrics;
  import org.apache.kafka.common.metrics.Sensor;
  import org.apache.kafka.common.metrics.stats.Avg;
  import org.apache.kafka.common.network.Selectable;
  import org.apache.kafka.common.protocol.ApiKeys;
  import org.apache.kafka.common.protocol.Errors;
  import org.apache.kafka.common.record.Record;
  import org.apache.kafka.common.record.RecordBatch;
  import org.apache.kafka.common.requests.AddOffsetsToTxnResponse;
  import org.apache.kafka.common.requests.EndTxnResponse;
  import org.apache.kafka.common.requests.FindCoordinatorRequest;
  import org.apache.kafka.common.requests.FindCoordinatorResponse;
  import org.apache.kafka.common.requests.InitProducerIdRequest;
  import org.apache.kafka.common.requests.InitProducerIdResponse;
  import org.apache.kafka.common.requests.JoinGroupRequest;
  import org.apache.kafka.common.requests.MetadataResponse;
  import org.apache.kafka.common.requests.ProduceResponse;
  import org.apache.kafka.common.requests.RequestTestUtils;
  import org.apache.kafka.common.requests.TxnOffsetCommitRequest;
  import org.apache.kafka.common.requests.TxnOffsetCommitResponse;
  import org.apache.kafka.common.serialization.ByteArraySerializer;
  import org.apache.kafka.common.serialization.Serializer;
  import org.apache.kafka.common.serialization.StringSerializer;
  import org.apache.kafka.common.telemetry.internals.ClientTelemetryReporter;
  import org.apache.kafka.common.telemetry.internals.ClientTelemetrySender;
  import org.apache.kafka.common.utils.LogCaptureAppender;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.common.utils.MockTime;
  import org.apache.kafka.common.utils.Time;
  import org.apache.kafka.test.MockMetricsReporter;
  import org.apache.kafka.test.MockPartitioner;
  import org.apache.kafka.test.MockProducerInterceptor;
  import org.apache.kafka.test.MockSerializer;
  import org.apache.kafka.test.TestUtils;
  
  import org.apache.logging.log4j.Level;
  import org.junit.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.TestInfo;
  import org.junit.jupiter.params.ParameterizedTest;
  import org.junit.jupiter.params.provider.CsvSource;
  import org.junit.jupiter.params.provider.ValueSource;
  import org.mockito.MockedStatic;
  import org.mockito.Mockito;
  import org.mockito.internal.stubbing.answers.CallsRealMethods;
  
  import java.lang.management.ManagementFactory;
  import java.nio.charset.StandardCharsets;
  import java.time.Duration;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.LinkedHashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Properties;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.Exchanger;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicReference;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  import java.util.stream.Stream;
  
  import javax.management.MBeanServer;
  import javax.management.ObjectName;
  
  import static java.util.Collections.emptyMap;
  import static java.util.Collections.singletonMap;
  import static org.apache.kafka.clients.producer.KafkaProducer.NETWORK_THREAD_PREFIX;
  import static org.junit.jupiter.api.Assertions.assertArrayEquals;
  import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertInstanceOf;
  import static org.junit.jupiter.api.Assertions.assertNotEquals;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertSame;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.junit.jupiter.api.Assertions.fail;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyInt;
  import static org.mockito.ArgumentMatchers.anyLong;
  import static org.mockito.ArgumentMatchers.anyString;
  import static org.mockito.ArgumentMatchers.eq;
  import static org.mockito.ArgumentMatchers.notNull;
  import static org.mockito.Mockito.atMostOnce;
  import static org.mockito.Mockito.doNothing;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.mockStatic;
  import static org.mockito.Mockito.never;
  import static org.mockito.Mockito.times;
  import static org.mockito.Mockito.verify;
  import static org.mockito.Mockito.when;
  
  public class KafkaProducerTest {
      private final String topic = "topic";
      private final Collection<Node> nodes = Collections.singletonList(NODE);
      private final Cluster emptyCluster = new Cluster(
              null,
              nodes,
              Collections.emptySet(),
              Collections.emptySet(),
              Collections.emptySet());
      private final Cluster onePartitionCluster = new Cluster(
              "dummy",
              nodes,
              Collections.singletonList(new PartitionInfo(topic, 0, null, null, null)),
              Collections.emptySet(),
              Collections.emptySet());
      private final Cluster threePartitionCluster = new Cluster(
              "dummy",
              nodes,
              Arrays.asList(
                      new PartitionInfo(topic, 0, null, null, null),
                      new PartitionInfo(topic, 1, null, null, null),
                      new PartitionInfo(topic, 2, null, null, null)),
              Collections.emptySet(),
              Collections.emptySet());
      private TestInfo testInfo;
  
      private static final int DEFAULT_METADATA_IDLE_MS = 5 * 60 * 1000;
      private static final Node NODE = new Node(0, "host1", 1000);
  
      private static <K, V> KafkaProducer<K, V> kafkaProducer(Map<String, Object> configs,
                    Serializer<K> keySerializer,
                    Serializer<V> valueSerializer,
                    ProducerMetadata metadata,
                    KafkaClient kafkaClient,
                    ProducerInterceptors<K, V> interceptors,
                    Time time) {
          return new KafkaProducer<>(new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, keySerializer, valueSerializer)),
              keySerializer, valueSerializer, metadata, kafkaClient, interceptors, new ApiVersions(), time);
      }
  
      @BeforeEach
      public void setup(TestInfo testInfo) {
          this.testInfo = testInfo;
      }
  
      @AfterEach
      public void detectLeaks() {
          // Assert no thread leakage of Kafka producer.
          TestUtils.assertNoLeakedThreadsWithNameAndDaemonStatus(NETWORK_THREAD_PREFIX, Boolean.TRUE);
      }
  
      @Test
      public void testOverwriteAcksAndRetriesForIdempotentProducers() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId");
          props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  
          ProducerConfig config = new ProducerConfig(props);
          assertTrue(config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG));
          assertTrue(Stream.of("-1", "all").anyMatch(each -> each.equalsIgnoreCase(config.getString(ProducerConfig.ACKS_CONFIG))));
          assertEquals((int) config.getInt(ProducerConfig.RETRIES_CONFIG), Integer.MAX_VALUE);
          assertTrue(config.getString(ProducerConfig.CLIENT_ID_CONFIG).equalsIgnoreCase("producer-" +
                  config.getString(ProducerConfig.TRANSACTIONAL_ID_CONFIG)));
      }
  
      @Test
      public void testAcksAndIdempotenceForIdempotentProducers() {
          Properties baseProps = baseProperties();
  
          Properties validProps = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.ACKS_CONFIG, "0");
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
              }};
          ProducerConfig config = new ProducerConfig(validProps);
          assertFalse(
              config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
              "idempotence should be overwritten");
          assertEquals(
              "0",
              config.getString(ProducerConfig.ACKS_CONFIG),
              "acks should be overwritten");
  
          Properties validProps2 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId");
              }};
          config = new ProducerConfig(validProps2);
          assertTrue(
              config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
              "idempotence should be set with the default value");
          assertEquals(
              "-1",
              config.getString(ProducerConfig.ACKS_CONFIG),
              "acks should be set with the default value");
  
          Properties validProps3 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.ACKS_CONFIG, "all");
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
              }};
          config = new ProducerConfig(validProps3);
          assertFalse(config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
              "idempotence should be overwritten");
          assertEquals(
              "-1",
              config.getString(ProducerConfig.ACKS_CONFIG),
              "acks should be overwritten");
  
          Properties validProps4 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.ACKS_CONFIG, "0");
              }};
          config = new ProducerConfig(validProps4);
          assertFalse(
              config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
              "idempotence should be disabled when acks not set to all and `enable.idempotence` config is unset.");
          assertEquals(
              "0",
              config.getString(ProducerConfig.ACKS_CONFIG),
              "acks should be set with overridden value");
  
          Properties validProps5 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.ACKS_CONFIG, "1");
              }};
          config = new ProducerConfig(validProps5);
          assertFalse(
              config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
              "idempotence should be disabled when acks not set to all and `enable.idempotence` config is unset.");
          assertEquals(
              "1",
              config.getString(ProducerConfig.ACKS_CONFIG),
              "acks should be set with overridden value");
  
          Properties invalidProps = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.ACKS_CONFIG, "0");
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
                  setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps),
              "Cannot set a transactional.id without also enabling idempotence");
  
          Properties invalidProps2 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.ACKS_CONFIG, "1");
                  // explicitly enabling idempotence should still throw exception
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps2),
              "Must set acks to all in order to use the idempotent producer");
  
          Properties invalidProps3 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.ACKS_CONFIG, "0");
                  setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps3),
              "Must set acks to all when using the transactional producer.");
      }
  
      @Test
      public void testRetriesAndIdempotenceForIdempotentProducers() {
          Properties baseProps = baseProperties();
  
          Properties validProps = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.RETRIES_CONFIG, "0");
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
              }};
          ProducerConfig config = new ProducerConfig(validProps);
          assertFalse(
              config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
              "idempotence should be overwritten");
          assertEquals(
              0,
              config.getInt(ProducerConfig.RETRIES_CONFIG),
              "retries should be overwritten");
  
          Properties validProps2 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.RETRIES_CONFIG, "0");
              }};
          config = new ProducerConfig(validProps2);
          assertFalse(
              config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
              "idempotence should be disabled when retries set to 0 and `enable.idempotence` config is unset.");
          assertEquals(
              0,
              config.getInt(ProducerConfig.RETRIES_CONFIG),
              "retries should be set with overridden value");
  
          Properties invalidProps = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.RETRIES_CONFIG, "0");
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
                  setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps),
              "Cannot set a transactional.id without also enabling idempotence");
  
          Properties invalidProps2 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.RETRIES_CONFIG, "0");
                  // explicitly enabling idempotence should still throw exception
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps2),
              "Must set retries to non-zero when using the idempotent producer.");
  
          Properties invalidProps3 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.RETRIES_CONFIG, "0");
                  setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps3),
              "Must set retries to non-zero when using the transactional producer.");
      }
  
      private Properties baseProperties() {
          Properties baseProps = new Properties();
          baseProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          baseProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          baseProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          return baseProps;
      }
  
      @Test
      public void testInflightRequestsAndIdempotenceForIdempotentProducers() {
          Properties baseProps = baseProperties();
  
          Properties validProps = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "6");
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
              }};
          ProducerConfig config = new ProducerConfig(validProps);
          assertFalse(
              config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
              "idempotence should be overwritten");
          assertEquals(
              6,
              config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION),
              "max.in.flight.requests.per.connection should be overwritten");
  
          Properties invalidProps1 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "6");
              }};
  
          ConfigException configException = assertThrows(ConfigException.class, () -> new ProducerConfig(invalidProps1));
          assertEquals("To use the idempotent producer, " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION +
                       " must be set to at most 5. Current value is 6.", configException.getMessage());
  
          Properties invalidProps2 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
                  setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps2),
              "Cannot set a transactional.id without also enabling idempotence");
  
          Properties invalidProps3 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "6");
                  // explicitly enabling idempotence should still throw exception
                  setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps3),
              "Must set max.in.flight.requests.per.connection to at most 5 when using the idempotent producer.");
  
          Properties invalidProps4 = new Properties() {{
                  putAll(baseProps);
                  setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "6");
                  setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId");
              }};
          assertThrows(
              ConfigException.class,
              () -> new ProducerConfig(invalidProps4),
              "Must set retries to non-zero when using the idempotent producer.");
      }
  
      @Test
      public void testMetricsReporterAutoGeneratedClientId() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                  props, new StringSerializer(), new StringSerializer())) {
              assertEquals(2, producer.metrics.reporters().size());
  
              MockMetricsReporter mockMetricsReporter = (MockMetricsReporter) producer.metrics.reporters().stream()
                      .filter(reporter -> reporter instanceof MockMetricsReporter).findFirst().get();
              assertEquals(producer.getClientId(), mockMetricsReporter.clientId);
          }
      }
  
      @Test
      public void testDisableJmxAndClientTelemetryReporter() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, "");
          props.setProperty(ProducerConfig.ENABLE_METRICS_PUSH_CONFIG, "false");
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer())) {
              assertTrue(producer.metrics.reporters().isEmpty());
          }
      }
  
      @Test
      public void testExplicitlyOnlyEnableJmxReporter() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, "org.apache.kafka.common.metrics.JmxReporter");
          props.setProperty(ProducerConfig.ENABLE_METRICS_PUSH_CONFIG, "false");
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer())) {
              assertEquals(1, producer.metrics.reporters().size());
              assertInstanceOf(JmxReporter.class, producer.metrics.reporters().get(0));
          }
      }
  
      @Test
      public void testExplicitlyOnlyEnableClientTelemetryReporter() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, "");
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer())) {
              assertEquals(1, producer.metrics.reporters().size());
              assertInstanceOf(ClientTelemetryReporter.class, producer.metrics.reporters().get(0));
          }
      }
  
      @Test
      public void testConstructorWithSerializers() {
          Properties producerProps = new Properties();
          producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          new KafkaProducer<>(producerProps, new ByteArraySerializer(), new ByteArraySerializer()).close();
      }
  
      @Test
      public void testNoSerializerProvided() {
  
          Properties producerProps = new Properties();
          producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          assertThrows(ConfigException.class, () -> {
              try (KafkaProducer<?, ?> producer = new KafkaProducer<>(producerProps)) {
                  // KafkaProducer will be closed automatically after the block
              }
          });
  
          final Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          // Invalid value null for configuration key.serializer: must be non-null.
          assertThrows(ConfigException.class, () -> {
              try (KafkaProducer<String, String> producer = new KafkaProducer<>(configs)) {
                  // KafkaProducer will be closed automatically after the block
              }
          });
      }
  
      @Test
      public void testConstructorFailureCloseResource() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.CLIENT_ID_CONFIG, "testConstructorClose");
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "some.invalid.hostname.foo.bar.local:9999");
          props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
  
          final int oldInitCount = MockMetricsReporter.INIT_COUNT.get();
          final int oldCloseCount = MockMetricsReporter.CLOSE_COUNT.get();
          try (KafkaProducer<byte[], byte[]> ignored = new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer())) {
              fail("should have caught an exception and returned");
          } catch (KafkaException e) {
              assertEquals(oldInitCount + 1, MockMetricsReporter.INIT_COUNT.get());
              assertEquals(oldCloseCount + 1, MockMetricsReporter.CLOSE_COUNT.get());
              assertEquals("Failed to construct kafka producer", e.getMessage());
          }
      }
  
      @Test
      public void testConstructorWithNotStringKey() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.put(1, "not string key");
          ConfigException ce = assertThrows(
              ConfigException.class,
              () -> new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()));
          assertTrue(ce.getMessage().contains("not string key"), "Unexpected exception message: " + ce.getMessage());
      }
  
      @Test
      public void testConstructorWithInvalidMetricReporterClass() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, "an.invalid.class");
          KafkaException ce = assertThrows(
                  KafkaException.class,
                  () -> new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()));
          assertTrue(ce.getMessage().contains("Failed to construct kafka producer"), "Unexpected exception message: " + ce.getMessage());
          assertTrue(ce.getCause().getMessage().contains("Class an.invalid.class cannot be found"), "Unexpected cause: " + ce.getCause());
      }
  
      @Test
      public void testSerializerClose() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testConstructorClose");
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
          configs.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL);
          final int oldInitCount = MockSerializer.INIT_COUNT.get();
          final int oldCloseCount = MockSerializer.CLOSE_COUNT.get();
  
          try (var ignored = new KafkaProducer<>(configs, new MockSerializer(), new MockSerializer())) {
              assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
              assertEquals(oldCloseCount, MockSerializer.CLOSE_COUNT.get());
          }
  
          assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
          assertEquals(oldCloseCount + 2, MockSerializer.CLOSE_COUNT.get());
      }
  
      @Test
      public void testInterceptorConstructClose() {
          try {
              Properties props = new Properties();
              // test with client ID assigned by KafkaProducer
              props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              props.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MockProducerInterceptor.class.getName());
              props.setProperty(MockProducerInterceptor.APPEND_STRING_PROP, "something");
  
              try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                      props, new StringSerializer(), new StringSerializer())) {
                  assertEquals(1, MockProducerInterceptor.INIT_COUNT.get());
                  assertEquals(0, MockProducerInterceptor.CLOSE_COUNT.get());
  
                  // Cluster metadata will only be updated on calling onSend.
                  assertNull(MockProducerInterceptor.CLUSTER_META.get());
              }
              assertEquals(1, MockProducerInterceptor.INIT_COUNT.get());
              assertEquals(1, MockProducerInterceptor.CLOSE_COUNT.get());
          } finally {
              // cleanup since we are using mutable static variables in MockProducerInterceptor
              MockProducerInterceptor.resetCounters();
          }
      }
      @Test
      public void testInterceptorConstructorConfigurationWithExceptionShouldCloseRemainingInstances() {
          final int targetInterceptor = 3;
          try {
              Properties props = new Properties();
              props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              props.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, org.apache.kafka.test.MockProducerInterceptor.class.getName() + ", "
                      +  org.apache.kafka.test.MockProducerInterceptor.class.getName() + ", "
                      +  org.apache.kafka.test.MockProducerInterceptor.class.getName());
              props.setProperty(MockProducerInterceptor.APPEND_STRING_PROP, "something");
  
              MockProducerInterceptor.setThrowOnConfigExceptionThreshold(targetInterceptor);
  
              assertThrows(KafkaException.class, () ->
                  new KafkaProducer<>(props, new StringSerializer(), new StringSerializer())
              );
  
              assertEquals(3, MockProducerInterceptor.CONFIG_COUNT.get());
              assertEquals(3, MockProducerInterceptor.CLOSE_COUNT.get());
  
          } finally {
              MockProducerInterceptor.resetCounters();
          }
      }
      @Test
      public void testPartitionerClose() {
          try {
              Properties props = new Properties();
              props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              MockPartitioner.resetCounters();
              props.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, MockPartitioner.class.getName());
  
              try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                      props, new StringSerializer(), new StringSerializer())) {
                  assertEquals(1, MockPartitioner.INIT_COUNT.get());
                  assertEquals(0, MockPartitioner.CLOSE_COUNT.get());
              }
  
              assertEquals(1, MockPartitioner.INIT_COUNT.get());
              assertEquals(1, MockPartitioner.CLOSE_COUNT.get());
          } finally {
              // cleanup since we are using mutable static variables in MockPartitioner
              MockPartitioner.resetCounters();
          }
      }
  
      @Test
      public void shouldCloseProperlyAndThrowIfInterrupted() throws Exception {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MockPartitioner.class.getName());
          configs.put(ProducerConfig.BATCH_SIZE_CONFIG, "1");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          final Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time);
  
          ExecutorService executor = Executors.newSingleThreadExecutor();
          final AtomicReference<Exception> closeException = new AtomicReference<>();
          try {
              Future<?> future = executor.submit(() -> {
                  producer.send(new ProducerRecord<>("topic", "key", "value"));
                  try {
                      producer.close();
                      fail("Close should block and throw.");
                  } catch (Exception e) {
                      closeException.set(e);
                  }
              });
  
              // Close producer should not complete until send succeeds
              try {
                  future.get(100, TimeUnit.MILLISECONDS);
                  fail("Close completed without waiting for send");
              } catch (java.util.concurrent.TimeoutException expected) { /* ignore */ }
  
              // Ensure send has started
              client.waitForRequests(1, 1000);
  
              assertTrue(future.cancel(true), "Close terminated prematurely");
  
              TestUtils.waitForCondition(() -> closeException.get() != null,
                      "InterruptException did not occur within timeout.");
  
              assertInstanceOf(InterruptException.class, closeException.get(), "Expected exception not thrown " + closeException);
          } finally {
              executor.shutdownNow();
          }
  
      }
  
      @Test
      public void testOsDefaultSocketBufferSizes() {
          Map<String, Object> config = new HashMap<>();
          config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          config.put(ProducerConfig.SEND_BUFFER_CONFIG, Selectable.USE_DEFAULT_BUFFER_SIZE);
          config.put(ProducerConfig.RECEIVE_BUFFER_CONFIG, Selectable.USE_DEFAULT_BUFFER_SIZE);
          new KafkaProducer<>(config, new ByteArraySerializer(), new ByteArraySerializer()).close();
      }
  
      @Test
      public void testInvalidSocketSendBufferSize() {
          Map<String, Object> config = new HashMap<>();
          config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          config.put(ProducerConfig.SEND_BUFFER_CONFIG, -2);
          assertThrows(KafkaException.class, () -> new KafkaProducer<>(config, new ByteArraySerializer(), new ByteArraySerializer()));
      }
  
      @Test
      public void testInvalidSocketReceiveBufferSize() {
          Map<String, Object> config = new HashMap<>();
          config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          config.put(ProducerConfig.RECEIVE_BUFFER_CONFIG, -2);
          assertThrows(KafkaException.class, () -> new KafkaProducer<>(config, new ByteArraySerializer(), new ByteArraySerializer()));
      }
  
      private static KafkaProducer<String, String> producerWithOverrideNewSender(Map<String, Object> configs,
                                                                                 ProducerMetadata metadata) {
          return producerWithOverrideNewSender(configs, metadata, Time.SYSTEM);
      }
  
      private static KafkaProducer<String, String> producerWithOverrideNewSender(Map<String, Object> configs,
                                                                                 ProducerMetadata metadata,
                                                                                 Time time) {
          // let mockClient#leastLoadedNode return the node directly so that we can isolate Metadata calls from KafkaProducer for idempotent producer
          MockClient mockClient = new MockClient(Time.SYSTEM, metadata) {
              @Override
              public LeastLoadedNode leastLoadedNode(long now) {
                  return new LeastLoadedNode(NODE, true);
              }
          };
  
          return new KafkaProducer<>(
                  new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, new StringSerializer(), new StringSerializer())),
                  new StringSerializer(), new StringSerializer(), metadata, mockClient, null, new ApiVersions(), time) {
              @Override
              Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) {
                  // give Sender its own Metadata instance so that we can isolate Metadata calls from KafkaProducer
                  return super.newSender(logContext, kafkaClient, newMetadata(0, 0, 100_000));
              }
          };
      }
  
      @ParameterizedTest
      @ValueSource(booleans = {true, false})
      public void testMetadataFetch(boolean isIdempotenceEnabled) throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled);
  
          ProducerMetadata metadata = mock(ProducerMetadata.class);
  
          // Return empty cluster 4 times and cluster from then on
          when(metadata.fetch()).thenReturn(emptyCluster, emptyCluster, emptyCluster, emptyCluster, onePartitionCluster);
  
          KafkaProducer<String, String> producer = producerWithOverrideNewSender(configs, metadata);
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, "value");
          producer.send(record);
  
          // One request update for each empty cluster returned
          verify(metadata, times(4)).requestUpdateForTopic(topic);
          verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
          verify(metadata, times(5)).fetch();
  
          // Should not request update for subsequent `send`
          producer.send(record, null);
          verify(metadata, times(4)).requestUpdateForTopic(topic);
          verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
          verify(metadata, times(6)).fetch();
  
          // Should not request update for subsequent `partitionsFor`
          producer.partitionsFor(topic);
          verify(metadata, times(4)).requestUpdateForTopic(topic);
          verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
          verify(metadata, times(7)).fetch();
  
          producer.close(Duration.ofMillis(0));
      }
  
      @ParameterizedTest
      @ValueSource(booleans = {true, false})
      public void testMetadataExpiry(boolean isIdempotenceEnabled) throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled);
          ProducerMetadata metadata = mock(ProducerMetadata.class);
  
          when(metadata.fetch()).thenReturn(onePartitionCluster, emptyCluster, onePartitionCluster);
  
          KafkaProducer<String, String> producer = producerWithOverrideNewSender(configs, metadata);
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, "value");
          producer.send(record);
  
          // Verify the topic's metadata isn't requested since it's already present.
          verify(metadata, times(0)).requestUpdateForTopic(topic);
          verify(metadata, times(0)).awaitUpdate(anyInt(), anyLong());
          verify(metadata, times(1)).fetch();
  
          // The metadata has been expired. Verify the producer requests the topic's metadata.
          producer.send(record, null);
          verify(metadata, times(1)).requestUpdateForTopic(topic);
          verify(metadata, times(1)).awaitUpdate(anyInt(), anyLong());
          verify(metadata, times(3)).fetch();
  
          producer.close(Duration.ofMillis(0));
      }
  
      @ParameterizedTest
      @ValueSource(booleans = {true, false})
      public void testMetadataTimeoutWithMissingTopic(boolean isIdempotenceEnabled) throws Exception {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled);
  
          // Create a record for a not-yet-created topic
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, 2, null, "value");
          ProducerMetadata metadata = mock(ProducerMetadata.class);
  
          MockTime mockTime = new MockTime();
          AtomicInteger invocationCount = new AtomicInteger(0);
          when(metadata.fetch()).then(invocation -> {
              invocationCount.incrementAndGet();
              if (invocationCount.get() == 5) {
                  mockTime.setCurrentTimeMs(mockTime.milliseconds() + 70000);
              }
  
              return emptyCluster;
          });
  
          KafkaProducer<String, String> producer = producerWithOverrideNewSender(configs, metadata, mockTime);
  
          // Four request updates where the topic isn't present, at which point the timeout expires and a
          // TimeoutException is thrown
          // For idempotence enabled case, the first metadata.fetch will be called in Sender#maybeSendAndPollTransactionalRequest
          Future<RecordMetadata> future = producer.send(record);
          verify(metadata, times(4)).requestUpdateForTopic(topic);
          verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
          verify(metadata, times(5)).fetch();
          try {
              assertInstanceOf(TimeoutException.class, assertThrows(ExecutionException.class, future::get).getCause());
          } finally {
              producer.close(Duration.ofMillis(0));
          }
      }
  
      @ParameterizedTest
      @ValueSource(booleans = {true, false})
      public void testMetadataWithPartitionOutOfRange(boolean isIdempotenceEnabled) throws Exception {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled);
  
          // Create a record with a partition higher than the initial (outdated) partition range
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, 2, null, "value");
          ProducerMetadata metadata = mock(ProducerMetadata.class);
  
          MockTime mockTime = new MockTime();
  
          when(metadata.fetch()).thenReturn(onePartitionCluster, onePartitionCluster, threePartitionCluster);
  
          KafkaProducer<String, String> producer = producerWithOverrideNewSender(configs, metadata, mockTime);
          // One request update if metadata is available but outdated for the given record
          producer.send(record);
          verify(metadata, times(2)).requestUpdateForTopic(topic);
          verify(metadata, times(2)).awaitUpdate(anyInt(), anyLong());
          verify(metadata, times(3)).fetch();
  
          producer.close(Duration.ofMillis(0));
      }
  
      @ParameterizedTest
      @ValueSource(booleans = {true, false})
      public void testMetadataTimeoutWithPartitionOutOfRange(boolean isIdempotenceEnabled) throws Exception {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled);
  
          // Create a record with a partition higher than the initial (outdated) partition range
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, 2, null, "value");
          ProducerMetadata metadata = mock(ProducerMetadata.class);
  
          MockTime mockTime = new MockTime();
          AtomicInteger invocationCount = new AtomicInteger(0);
          when(metadata.fetch()).then(invocation -> {
              invocationCount.incrementAndGet();
              if (invocationCount.get() == 5) {
                  mockTime.setCurrentTimeMs(mockTime.milliseconds() + 70000);
              }
  
              return onePartitionCluster;
          });
  
          KafkaProducer<String, String> producer = producerWithOverrideNewSender(configs, metadata, mockTime);
  
          // Four request updates where the requested partition is out of range, at which point the timeout expires
          // and a TimeoutException is thrown
          // For idempotence enabled case, the first and last metadata.fetch will be called in Sender#maybeSendAndPollTransactionalRequest,
          // before the producer#send and after it finished
          Future<RecordMetadata> future = producer.send(record);
  
          verify(metadata, times(4)).requestUpdateForTopic(topic);
          verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
          verify(metadata, times(5)).fetch();
          try {
              assertInstanceOf(TimeoutException.class, assertThrows(ExecutionException.class, future::get).getCause());
          } finally {
              producer.close(Duration.ofMillis(0));
          }
      }
  
      @Test
      public void testTopicRefreshInMetadata() throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "600000");
          // test under normal producer for simplicity
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
          long refreshBackoffMs = 500L;
          long refreshBackoffMaxMs = 5000L;
          long metadataExpireMs = 60000L;
          long metadataIdleMs = 60000L;
          final Time time = new MockTime();
          final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs,
                  new LogContext(), new ClusterResourceListeners(), time);
          final String topic = "topic";
          try (KafkaProducer<String, String> producer = kafkaProducer(configs,
                  new StringSerializer(), new StringSerializer(), metadata, new MockClient(time, metadata), null, time)) {
  
              AtomicBoolean running = new AtomicBoolean(true);
              Thread t = new Thread(() -> {
                  long startTimeMs = System.currentTimeMillis();
                  while (running.get()) {
                      while (!metadata.updateRequested() && System.currentTimeMillis() - startTimeMs < 100)
                          Thread.yield();
                      MetadataResponse updateResponse = RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                              singletonMap(topic, Errors.UNKNOWN_TOPIC_OR_PARTITION), emptyMap());
                      metadata.updateWithCurrentRequestVersion(updateResponse, false, time.milliseconds());
                      time.sleep(60 * 1000L);
                  }
              });
              t.start();
              Throwable throwable = assertThrows(TimeoutException.class, () -> producer.partitionsFor(topic));
              assertInstanceOf(UnknownTopicOrPartitionException.class, throwable.getCause());
              running.set(false);
              t.join();
          }
      }
  
      @Test
      public void testTopicNotExistingInMetadata() throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "30000");
          long refreshBackoffMs = 500L;
          long refreshBackoffMaxMs = 5000L;
          long metadataExpireMs = 60000L;
          long metadataIdleMs = 60000L;
          final Time time = new MockTime();
          final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs,
                  new LogContext(), new ClusterResourceListeners(), time);
          final String topic = "topic";
          try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, new MockClient(time, metadata), null, time)) {
  
              Exchanger<Void> exchanger = new Exchanger<>();
  
              Thread t = new Thread(() -> {
                  try {
                      // Update the metadata with non-existing topic.
                      MetadataResponse updateResponse = RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                              singletonMap(topic, Errors.UNKNOWN_TOPIC_OR_PARTITION), emptyMap());
                      metadata.updateWithCurrentRequestVersion(updateResponse, false, time.milliseconds());
                      exchanger.exchange(null);
                      while (!metadata.updateRequested())
                          Thread.sleep(100);
                      time.sleep(30 * 1000L);
                  } catch (Exception e) {
                      throw new RuntimeException(e);
                  }
              });
              t.start();
              exchanger.exchange(null);
              Throwable throwable = assertThrows(TimeoutException.class, () -> producer.partitionsFor(topic));
              assertInstanceOf(UnknownTopicOrPartitionException.class, throwable.getCause());
              t.join();
          }
      }
  
      @Test
      public void testTopicExpiryInMetadata() throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "30000");
          long refreshBackoffMs = 500L;
          long refreshBackoffMaxMs = 5000L;
          long metadataExpireMs = 60000L;
          long metadataIdleMs = 60000L;
          final Time time = new MockTime();
          final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs,
                  new LogContext(), new ClusterResourceListeners(), time);
          final String topic = "topic";
          try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, new MockClient(time, metadata), null, time)) {
  
              Exchanger<Void> exchanger = new Exchanger<>();
  
              Thread t = new Thread(() -> {
                  try {
                      exchanger.exchange(null);  // 1
                      while (!metadata.updateRequested())
                          Thread.sleep(100);
                      MetadataResponse updateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1));
                      metadata.updateWithCurrentRequestVersion(updateResponse, false, time.milliseconds());
                      exchanger.exchange(null);  // 2
                      time.sleep(120 * 1000L);
  
                      // Update the metadata again, but it should be expired at this point.
                      updateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1));
                      metadata.updateWithCurrentRequestVersion(updateResponse, false, time.milliseconds());
                      exchanger.exchange(null);  // 3
                      while (!metadata.updateRequested())
                          Thread.sleep(100);
                      time.sleep(30 * 1000L);
                  } catch (Exception e) {
                      throw new RuntimeException(e);
                  }
              });
              t.start();
              exchanger.exchange(null);  // 1
              assertNotNull(producer.partitionsFor(topic));
              exchanger.exchange(null);  // 2
              exchanger.exchange(null);  // 3
              assertThrows(TimeoutException.class, () -> producer.partitionsFor(topic));
              t.join();
          }
      }
  
      @SuppressWarnings("unchecked")
      @Test
      public void testHeadersSuccess() {
          doTestHeaders(Serializer.class);
      }
  
      private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerClassToMock) {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorForHeaders.class.getName());
          Serializer<String> keySerializer = mock(serializerClassToMock);
          Serializer<String> valueSerializer = mock(serializerClassToMock);
  
          long nowMs = Time.SYSTEM.milliseconds();
          String topic = "topic";
          ProducerMetadata metadata = newMetadata(0, 0, 90000);
          metadata.add(topic, nowMs);
  
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1));
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, nowMs);
  
          KafkaProducer<String, String> producer = kafkaProducer(configs, keySerializer, valueSerializer, metadata,
                  null, null, Time.SYSTEM);
  
          when(keySerializer.serialize(any(), any(), any())).then(invocation ->
                  invocation.<String>getArgument(2).getBytes());
          when(valueSerializer.serialize(any(), any(), any())).then(invocation ->
                  invocation.<String>getArgument(2).getBytes());
  
          String value = "value";
          String key = "key";
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
  
          //ensure headers can be mutated pre send.
          record.headers().add(new RecordHeader("test", "header2".getBytes()));
          producer.send(record, null);
  
          //ensure headers are closed and cannot be mutated post send
          RecordHeaders recordHeaders = (RecordHeaders) record.headers();
          assertTrue(recordHeaders.isReadOnly());
          assertThrows(IllegalStateException.class, () -> recordHeaders.add(new RecordHeader("test", "test".getBytes())));
  
          //ensure existing headers are not changed, and last header for key is still original value
          assertArrayEquals(record.headers().lastHeader("test").value(), "header2".getBytes());
  
          verify(valueSerializer).serialize(topic, record.headers(), value);
          verify(keySerializer).serialize(topic, record.headers(), key);
  
          producer.close(Duration.ofMillis(0));
      }
  
      @Test
      public void testHeadersFailure() {
          Properties props = new Properties();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5);
          props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorForHeaders.class.getName());
          Serializer<String> keySerializer = mock(StringSerializer.class);
          Serializer<String> valueSerializer = mock(StringSerializer.class);
  
          KafkaProducer<String, String> producer = new KafkaProducer<>(props, keySerializer, valueSerializer);
          ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
          Future<RecordMetadata> future = producer.send(record, (recordMetadata, exception) -> { });
          try {
              TestUtils.assertFutureThrows(TimeoutException.class, future);
              //ensure headers are writable if send failure
              RecordHeaders recordHeaders = (RecordHeaders) record.headers();
              assertFalse(recordHeaders.isReadOnly());
          } finally {
              producer.close(Duration.ofMillis(0));
          }
      }
  
      @Test
      public void closeShouldBeIdempotent() {
          Properties producerProps = new Properties();
          producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          Producer<byte[], byte[]> producer = new KafkaProducer<>(producerProps, new ByteArraySerializer(), new ByteArraySerializer());
          producer.close();
          producer.close();
      }
  
      @Test
      public void closeWithNegativeTimestampShouldThrow() {
          Properties producerProps = new Properties();
          producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          try (Producer<byte[], byte[]> producer = new KafkaProducer<>(producerProps, new ByteArraySerializer(), new ByteArraySerializer())) {
              assertThrows(IllegalArgumentException.class, () -> producer.close(Duration.ofMillis(-100)));
          }
      }
  
      @Test
      public void testFlushCompleteSendOfInflightBatches() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          // only test in idempotence disabled producer for simplicity
          // flush operation acts the same for idempotence enabled and disabled cases
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              ArrayList<Future<RecordMetadata>> futureResponses = new ArrayList<>();
              for (int i = 0; i < 50; i++) {
                  Future<RecordMetadata> response = producer.send(new ProducerRecord<>("topic", "value" + i));
                  futureResponses.add(response);
              }
  
              futureResponses.forEach(res -> assertFalse(res.isDone()));
              producer.flush();
              futureResponses.forEach(res -> assertTrue(res.isDone()));
          }
      }
  
      private static Double getMetricValue(final KafkaProducer<?, ?> producer, final String name) {
          Metrics metrics = producer.metrics;
          Metric metric =  metrics.metric(metrics.metricName(name, "producer-metrics"));
          return (Double) metric.metricValue();
      }
  
      @Test
      public void testFlushMeasureLatency() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          try (KafkaProducer<String, String> producer = kafkaProducer(
              configs,
              new StringSerializer(),
              new StringSerializer(),
              metadata,
              client,
              null,
              time
          )) {
              producer.flush();
              double first = getMetricValue(producer, "flush-time-ns-total");
              assertTrue(first > 0);
              producer.flush();
              assertTrue(getMetricValue(producer, "flush-time-ns-total") > first);
          }
      }
  
      @Test
      public void testMetricConfigRecordingLevel() {
          Properties props = new Properties();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer())) {
              assertEquals(Sensor.RecordingLevel.INFO, producer.metrics.config().recordLevel());
          }
  
          props.put(ProducerConfig.METRICS_RECORDING_LEVEL_CONFIG, "DEBUG");
          try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer())) {
              assertEquals(Sensor.RecordingLevel.DEBUG, producer.metrics.config().recordLevel());
          }
      }
  
      @Test
      public void testInterceptorPartitionSetOnTooLargeRecord() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "1");
          String topic = "topic";
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, "value");
  
          long nowMs = Time.SYSTEM.milliseconds();
          ProducerMetadata metadata = newMetadata(0, 0, 90000);
          metadata.add(topic, nowMs);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1));
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, nowMs);
  
          @SuppressWarnings("unchecked") // it is safe to suppress, since this is a mock class
                  ProducerInterceptors<String, String> interceptors = mock(ProducerInterceptors.class);
          KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, null, interceptors, Time.SYSTEM);
  
          when(interceptors.onSend(any())).then(invocation -> invocation.getArgument(0));
  
          producer.send(record);
  
          verify(interceptors).onSend(record);
          verify(interceptors).onSendError(eq(record), notNull(), notNull());
  
          producer.close(Duration.ofMillis(0));
      }
  
      @Test
      public void testPartitionsForWithNullTopic() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer())) {
              assertThrows(NullPointerException.class, () -> producer.partitionsFor(null));
          }
      }
  
      @Test
      public void testInitTransactionsResponseAfterTimeout() throws Exception {
          int maxBlockMs = 500;
  
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
  
          ExecutorService executor = Executors.newFixedThreadPool(1);
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              client.prepareResponse(
                  request -> request instanceof FindCoordinatorRequest &&
                      ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(),
                  FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", NODE));
  
              Future<?> future = executor.submit(() -> producer.initTransactions());
              TestUtils.waitForCondition(client::hasInFlightRequests,
                  "Timed out while waiting for expected `InitProducerId` request to be sent");
  
              time.sleep(maxBlockMs);
              TestUtils.assertFutureThrows(TimeoutException.class, future);
  
              client.respond(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
              Thread.sleep(1000);
              producer.initTransactions();
          }
      }
  
      @Test
      public void testInitTransactionTimeout() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 500);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              client.prepareResponse(
                  request -> request instanceof FindCoordinatorRequest &&
                      ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(),
                  FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", NODE));
  
              assertThrows(TimeoutException.class, producer::initTransactions);
  
              client.prepareResponse(
                  request -> request instanceof FindCoordinatorRequest &&
                                 ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(),
                  FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", NODE));
  
              client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
              // retry initialization should work
              producer.initTransactions();
          }
      }
  
      @Test
      public void testInitTransactionWhileThrottled() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          Node node = metadata.fetch().nodes().get(0);
          client.throttle(node, 5000);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
          }
      }
  
      @ParameterizedTest
      @CsvSource({
          "true, false",
          "true, true",
          "false, true"
      })
      public void testInitTransactionsWithKeepPreparedTxnAndTwoPhaseCommit(boolean keepPreparedTxn, boolean enable2PC) {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-txn-id");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          if (enable2PC) {
              configs.put(ProducerConfig.TRANSACTION_TWO_PHASE_COMMIT_ENABLE_CONFIG, true);
          }
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          // Capture flags from the InitProducerIdRequest
          boolean[] requestFlags = new boolean[2]; // [keepPreparedTxn, enable2Pc]
          
          client.prepareResponse(
              request -> request instanceof FindCoordinatorRequest &&
                  ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(),
              FindCoordinatorResponse.prepareResponse(Errors.NONE, "test-txn-id", NODE));
              
          client.prepareResponse(
              request -> {
                  if (request instanceof InitProducerIdRequest) {
                      InitProducerIdRequest initRequest = (InitProducerIdRequest) request;
                      requestFlags[0] = initRequest.data().keepPreparedTxn();
                      requestFlags[1] = initRequest.data().enable2Pc();
                      return true;
                  }
                  return false;
              },
              initProducerIdResponse(1L, (short) 5, Errors.NONE));
              
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions(keepPreparedTxn);
              
              // Verify request flags match expected values
              assertEquals(keepPreparedTxn, requestFlags[0], 
                  "keepPreparedTxn flag should match input parameter");
              assertEquals(enable2PC, requestFlags[1], 
                  "enable2Pc flag should match producer configuration");
          }
      }
  
      @Test
      public void testPrepareTransactionSuccess() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          when(ctx.transactionManager.isTransactionV2Enabled()).thenReturn(true);
          when(ctx.transactionManager.is2PCEnabled()).thenReturn(true);
          when(ctx.sender.isRunning()).thenReturn(true);
  
          doNothing().when(ctx.transactionManager).prepareTransaction();
  
          PreparedTxnState expectedState = mock(PreparedTxnState.class);
          when(ctx.transactionManager.preparedTransactionState()).thenReturn(expectedState);
  
          try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
              PreparedTxnState returned = producer.prepareTransaction();
              assertSame(expectedState, returned);
  
              verify(ctx.transactionManager).prepareTransaction();
              verify(ctx.accumulator).beginFlush();
              verify(ctx.accumulator).awaitFlushCompletion();
          }
      }
  
      @Test
      public void testSendNotAllowedInPreparedTransactionState() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          String topic = "foo";
          Cluster cluster = TestUtils.singletonCluster(topic, 1);
  
          when(ctx.sender.isRunning()).thenReturn(true);
          when(ctx.metadata.fetch()).thenReturn(cluster);
  
          // Mock transaction manager to simulate being in a prepared state
          when(ctx.transactionManager.isTransactional()).thenReturn(true);
          when(ctx.transactionManager.isPrepared()).thenReturn(true);
  
          // Create record to send
          long timestamp = ctx.time.milliseconds();
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, 0, timestamp, "key", "value");
  
          try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
              // Verify that sending a record throws IllegalStateException with the correct message
              IllegalStateException exception = assertThrows(
                  IllegalStateException.class,
                  () -> producer.send(record)
              );
  
              assertTrue(exception.getMessage().contains("Cannot perform operation while the transaction is in a prepared state"));
  
              // Verify transactionManager methods were called
              verify(ctx.transactionManager).isTransactional();
              verify(ctx.transactionManager).isPrepared();
  
              // Verify that no message was actually sent (accumulator was not called)
              verify(ctx.accumulator, never()).append(
                  eq(topic),
                  anyInt(),
                  anyLong(),
                  any(),
                  any(),
                  any(),
                  any(),
                  anyLong(),
                  anyLong(),
                  any()
              );
          }
      }
  
      @Test
      public void testSendOffsetsNotAllowedInPreparedTransactionState() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          String topic = "foo";
          Cluster cluster = TestUtils.singletonCluster(topic, 1);
  
          when(ctx.sender.isRunning()).thenReturn(true);
          when(ctx.metadata.fetch()).thenReturn(cluster);
  
          // Mock transaction manager to simulate being in a prepared state
          when(ctx.transactionManager.isTransactional()).thenReturn(true);
          when(ctx.transactionManager.isPrepared()).thenReturn(true);
  
          // Create consumer group metadata
          String groupId = "test-group";
          Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
          offsets.put(new TopicPartition(topic, 0), new OffsetAndMetadata(100L));
          ConsumerGroupMetadata groupMetadata = new ConsumerGroupMetadata(groupId);
  
          try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
              // Verify that sending offsets throws IllegalStateException with the correct message
              IllegalStateException exception = assertThrows(
                  IllegalStateException.class,
                  () -> producer.sendOffsetsToTransaction(offsets, groupMetadata)
              );
  
              assertTrue(exception.getMessage().contains("Cannot perform operation while the transaction is in a prepared state"));
  
              // Verify transactionManager methods were called
              verify(ctx.transactionManager).isTransactional();
              verify(ctx.transactionManager).isPrepared();
  
              // Verify that no offsets were actually sent
              verify(ctx.transactionManager, never()).sendOffsetsToTransaction(
                  eq(offsets),
                  eq(groupMetadata)
              );
          }
      }
  
      @Test
      public void testBeginTransactionNotAllowedInPreparedTransactionState() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          when(ctx.sender.isRunning()).thenReturn(true);
  
          // Mock transaction manager to simulate being in a prepared state
          when(ctx.transactionManager.isTransactional()).thenReturn(true);
          when(ctx.transactionManager.isPrepared()).thenReturn(true);
  
          try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
              // Verify that calling beginTransaction throws IllegalStateException with the correct message
              IllegalStateException exception = assertThrows(
                  IllegalStateException.class,
                  producer::beginTransaction
              );
  
              assertTrue(exception.getMessage().contains("Cannot perform operation while the transaction is in a prepared state"));
  
              // Verify transactionManager methods were called
              verify(ctx.transactionManager).isTransactional();
              verify(ctx.transactionManager).isPrepared();
          }
      }
  
      @Test
      public void testPrepareTransactionFailsWhen2PCDisabled() {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          // Disable 2PC
          when(ctx.transactionManager.isTransactionV2Enabled()).thenReturn(true);
          when(ctx.transactionManager.is2PCEnabled()).thenReturn(false);
          when(ctx.sender.isRunning()).thenReturn(true);
  
          try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
              assertThrows(
                  InvalidTxnStateException.class,
                  producer::prepareTransaction,
                  "prepareTransaction() should fail if 2PC is disabled"
              );
          }
      }
      
      @Test
      public void testCompleteTransactionWithMatchingState() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          when(ctx.transactionManager.isPrepared()).thenReturn(true);
          when(ctx.sender.isRunning()).thenReturn(true);
          
          // Create prepared states with matching values
          long producerId = 12345L;
          short epoch = 5;
          PreparedTxnState currentState = new PreparedTxnState(producerId, epoch);
          PreparedTxnState inputState = new PreparedTxnState(producerId, epoch);
          
          // Set up the transaction manager to return the prepared state
          when(ctx.transactionManager.preparedTransactionState()).thenReturn(currentState);
          
          // Should trigger commit when states match
          TransactionalRequestResult commitResult = mock(TransactionalRequestResult.class);
          when(ctx.transactionManager.beginCommit()).thenReturn(commitResult);
          
          try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
              // Call completeTransaction with the matching state
              producer.completeTransaction(inputState);
              
              // Verify methods called in order
              verify(ctx.transactionManager).isPrepared();
              verify(ctx.transactionManager).preparedTransactionState();
              verify(ctx.transactionManager).beginCommit();
              
              // Verify abort was never called
              verify(ctx.transactionManager, never()).beginAbort();
              
              // Verify sender was woken up
              verify(ctx.sender).wakeup();
          }
      }
      
      @Test
      public void testCompleteTransactionWithNonMatchingState() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          when(ctx.transactionManager.isPrepared()).thenReturn(true);
          when(ctx.sender.isRunning()).thenReturn(true);
          
          // Create txn prepared states with different values
          long producerId = 12345L;
          short epoch = 5;
          PreparedTxnState currentState = new PreparedTxnState(producerId, epoch);
          PreparedTxnState inputState = new PreparedTxnState(producerId + 1, epoch);
          
          // Set up the transaction manager to return the prepared state
          when(ctx.transactionManager.preparedTransactionState()).thenReturn(currentState);
          
          // Should trigger abort when states don't match
          TransactionalRequestResult abortResult = mock(TransactionalRequestResult.class);
          when(ctx.transactionManager.beginAbort()).thenReturn(abortResult);
          
          try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
              // Call completeTransaction with the non-matching state
              producer.completeTransaction(inputState);
              
              // Verify methods called in order
              verify(ctx.transactionManager).isPrepared();
              verify(ctx.transactionManager).preparedTransactionState();
              verify(ctx.transactionManager).beginAbort();
              
              // Verify commit was never called
              verify(ctx.transactionManager, never()).beginCommit();
              
              // Verify sender was woken up
              verify(ctx.sender).wakeup();
          }
      }
      
      @Test
      public void testClusterAuthorizationFailure() throws Exception {
          int maxBlockMs = 500;
  
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some-txn");
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(500, 5000, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some-txn", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.CLUSTER_AUTHORIZATION_FAILED));
          Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time);
          assertThrows(ClusterAuthorizationException.class, producer::initTransactions);
  
          // retry initTransactions after the ClusterAuthorizationException not being thrown
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
          TestUtils.retryOnExceptionWithTimeout(1000, 100, producer::initTransactions);
          producer.close();
      }
  
      @Test
      public void testAbortTransaction() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
          client.prepareResponse(endTxnResponse(Errors.NONE));
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
              producer.beginTransaction();
              producer.abortTransaction();
          }
      }
  
      @Test
      public void testTransactionV2ProduceWithConcurrentTransactionError() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          String topic = "foo";
          TopicIdPartition topicIdPartition = new TopicIdPartition(Uuid.randomUuid(), 0, topic);
          Cluster cluster = TestUtils.singletonCluster(topic, 1);
  
          when(ctx.sender.isRunning()).thenReturn(true);
          when(ctx.metadata.fetch()).thenReturn(cluster);
  
          long timestamp = ctx.time.milliseconds();
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, 0, timestamp, "key", "value");
  
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some-txn");
          props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          ProducerConfig config = new ProducerConfig(props);
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
          NodeApiVersions nodeApiVersions = new NodeApiVersions(NodeApiVersions.create().allSupportedApiVersions().values(),
              Arrays.asList(new ApiVersionsResponseData.SupportedFeatureKey()
                  .setName("transaction.version")
                  .setMaxVersion((short) 2)
                  .setMinVersion((short) 0)),
              Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
                  .setName("transaction.version")
                  .setMaxVersionLevel((short) 2)
                  .setMinVersionLevel((short) 2)),
              0);
          client.setNodeApiVersions(nodeApiVersions);
          ApiVersions apiVersions = new ApiVersions();
          apiVersions.update(NODE.idString(), nodeApiVersions);
  
          ProducerInterceptors<String, String> interceptor = new ProducerInterceptors<>(Collections.emptyList(), null);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some-txn", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
          client.prepareResponse(produceResponse(topicIdPartition, 1L, Errors.CONCURRENT_TRANSACTIONS, 0, 1));
          client.prepareResponse(produceResponse(topicIdPartition, 1L, Errors.NONE, 0, 1));
          client.prepareResponse(endTxnResponse(Errors.NONE));
  
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(
              config, new StringSerializer(), new StringSerializer(), metadata, client, interceptor, apiVersions, time)
          ) {
              producer.initTransactions();
              producer.beginTransaction();
              producer.send(record).get();
              producer.commitTransaction();
          }
      }
  
      @Test
      public void testMeasureAbortTransactionDuration() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
          try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
              new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
  
              client.prepareResponse(endTxnResponse(Errors.NONE));
              producer.beginTransaction();
              producer.abortTransaction();
              double first = getMetricValue(producer, "txn-abort-time-ns-total");
              assertTrue(first > 0);
  
              client.prepareResponse(endTxnResponse(Errors.NONE));
              producer.beginTransaction();
              producer.abortTransaction();
              assertTrue(getMetricValue(producer, "txn-abort-time-ns-total") > first);
          }
      }
  
      @Test
      public void testCommitTransactionWithRecordTooLargeException() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1000);
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = mock(ProducerMetadata.class);
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
          when(metadata.fetch()).thenReturn(onePartitionCluster);
  
          String largeString = IntStream.range(0, 1000).mapToObj(i -> "*").collect(Collectors.joining());
          ProducerRecord<String, String> largeRecord = new ProducerRecord<>(topic, "large string", largeString);
  
          try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
  
              client.prepareResponse(endTxnResponse(Errors.NONE));
              producer.beginTransaction();
              TestUtils.assertFutureThrows(RecordTooLargeException.class, producer.send(largeRecord));
              assertThrows(KafkaException.class, producer::commitTransaction);
          }
      }
  
      @Test
      public void testCommitTransactionWithMetadataTimeoutForMissingTopic() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
  
          // Create a record for a not-yet-created topic
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, "value");
          ProducerMetadata metadata = mock(ProducerMetadata.class);
  
          MockTime mockTime = new MockTime();
  
          MockClient client = new MockClient(mockTime, metadata);
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
          AtomicInteger invocationCount = new AtomicInteger(0);
          when(metadata.fetch()).then(invocation -> {
              invocationCount.incrementAndGet();
              if (invocationCount.get() > 5) {
                  mockTime.setCurrentTimeMs(mockTime.milliseconds() + 70000);
              }
  
              return emptyCluster;
          });
  
          try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, mockTime)) {
              producer.initTransactions();
              producer.beginTransaction();
  
              TestUtils.assertFutureThrows(TimeoutException.class, producer.send(record));
              assertThrows(KafkaException.class, producer::commitTransaction);
          }
      }
  
      @Test
      public void testCommitTransactionWithMetadataTimeoutForPartitionOutOfRange() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
  
          // Create a record with a partition higher than the initial (outdated) partition range
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, 2, null, "value");
          ProducerMetadata metadata = mock(ProducerMetadata.class);
  
          MockTime mockTime = new MockTime();
  
          MockClient client = new MockClient(mockTime, metadata);
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
          AtomicInteger invocationCount = new AtomicInteger(0);
          when(metadata.fetch()).then(invocation -> {
              invocationCount.incrementAndGet();
              if (invocationCount.get() > 5) {
                  mockTime.setCurrentTimeMs(mockTime.milliseconds() + 70000);
              }
  
              return onePartitionCluster;
          });
  
          try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, mockTime)) {
              producer.initTransactions();
              producer.beginTransaction();
  
              TestUtils.assertFutureThrows(TimeoutException.class, producer.send(record));
              assertThrows(KafkaException.class, producer::commitTransaction);
          }
      }
  
      @Test
      public void testCommitTransactionWithSendToInvalidTopic() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "15000");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, emptyMap());
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
          String invalidTopicName = "topic abc"; // Invalid topic name due to space
          ProducerRecord<String, String> record = new ProducerRecord<>(invalidTopicName, "HelloKafka");
  
          List<MetadataResponse.TopicMetadata> topicMetadata = new ArrayList<>();
          topicMetadata.add(new MetadataResponse.TopicMetadata(Errors.INVALID_TOPIC_EXCEPTION,
                  invalidTopicName, false, Collections.emptyList()));
          MetadataResponse updateResponse =  RequestTestUtils.metadataResponse(
                  new ArrayList<>(initialUpdateResponse.brokers()),
                  initialUpdateResponse.clusterId(),
                  initialUpdateResponse.controller().id(),
                  topicMetadata);
          client.prepareMetadataUpdate(updateResponse);
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
              producer.beginTransaction();
  
              TestUtils.assertFutureThrows(InvalidTopicException.class, producer.send(record));
              assertThrows(KafkaException.class, producer::commitTransaction);
          }
      }
  
      @Test
      public void testSendTxnOffsetsWithGroupId() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          Node node = metadata.fetch().nodes().get(0);
          client.throttle(node, 5000);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
          client.prepareResponse(addOffsetsToTxnResponse(Errors.NONE));
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          String groupId = "group";
          client.prepareResponse(request ->
              ((TxnOffsetCommitRequest) request).data().groupId().equals(groupId),
              txnOffsetsCommitResponse(Collections.singletonMap(
                  new TopicPartition("topic", 0), Errors.NONE)));
          client.prepareResponse(endTxnResponse(Errors.NONE));
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
              new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
              producer.beginTransaction();
              producer.sendOffsetsToTransaction(Collections.emptyMap(), new ConsumerGroupMetadata(groupId));
              producer.commitTransaction();
          }
      }
  
      @Test
      public void testSendTxnOffsetsWithGroupIdTransactionV2() {
          Properties properties = new Properties();
          properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
          properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          Node node = metadata.fetch().nodes().get(0);
          client.setNodeApiVersions(NodeApiVersions.create());
          NodeApiVersions nodeApiVersions = new NodeApiVersions(NodeApiVersions.create().allSupportedApiVersions().values(),
              Arrays.asList(new ApiVersionsResponseData.SupportedFeatureKey()
                  .setName("transaction.version")
                  .setMaxVersion((short) 2)
                  .setMinVersion((short) 0)),
              Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
                  .setName("transaction.version")
                  .setMaxVersionLevel((short) 2)
                  .setMinVersionLevel((short) 2)),
              0);
          client.setNodeApiVersions(nodeApiVersions);
          ApiVersions apiVersions = new ApiVersions();
          apiVersions.update(NODE.idString(), nodeApiVersions);
  
          client.throttle(node, 5000);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          String groupId = "group";
          client.prepareResponse(request ->
              ((TxnOffsetCommitRequest) request).data().groupId().equals(groupId),
              txnOffsetsCommitResponse(Collections.singletonMap(
                  new TopicPartition("topic", 0), Errors.NONE)));
          client.prepareResponse(endTxnResponse(Errors.NONE));
  
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(
              new ProducerConfig(properties), new StringSerializer(), new StringSerializer(), metadata, client,
              new ProducerInterceptors<>(Collections.emptyList(), null), apiVersions, time)) {
              producer.initTransactions();
              producer.beginTransaction();
              producer.sendOffsetsToTransaction(Collections.singletonMap(
                  new TopicPartition("topic", 0),
                  new OffsetAndMetadata(5L)),
                  new ConsumerGroupMetadata(groupId));
              producer.commitTransaction();
          }
      }
  
      @Test
      public void testTransactionV2Produce() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          String topic = "foo";
          Uuid topicId = Uuid.fromString("klZ9sa2rSvig6QpgGXzALT");
          TopicPartition topicPartition = new TopicPartition(topic, 0);
          Cluster cluster = TestUtils.singletonCluster(topic, 1);
  
          when(ctx.sender.isRunning()).thenReturn(true);
          when(ctx.metadata.fetch()).thenReturn(cluster);
  
          long timestamp = ctx.time.milliseconds();
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, 0, timestamp, "key", "value");
  
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some-txn");
          props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          ProducerConfig config = new ProducerConfig(props);
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
          NodeApiVersions nodeApiVersions = new NodeApiVersions(NodeApiVersions.create().allSupportedApiVersions().values(),
              Arrays.asList(new ApiVersionsResponseData.SupportedFeatureKey()
                  .setName("transaction.version")
                  .setMaxVersion((short) 2)
                  .setMinVersion((short) 0)),
              Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
                  .setName("transaction.version")
                  .setMaxVersionLevel((short) 2)
                  .setMinVersionLevel((short) 2)),
              0);
          client.setNodeApiVersions(nodeApiVersions);
          ApiVersions apiVersions = new ApiVersions();
          apiVersions.update(NODE.idString(), nodeApiVersions);
  
          ProducerInterceptors<String, String> interceptor = new ProducerInterceptors<>(Collections.emptyList(), null);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some-txn", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
          client.prepareResponse(produceResponse(new TopicIdPartition(topicId, topicPartition), 1L, Errors.NONE, 0, 1));
          client.prepareResponse(endTxnResponse(Errors.NONE));
  
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                  config, new StringSerializer(), new StringSerializer(), metadata, client, interceptor, apiVersions, time)
          ) {
              producer.initTransactions();
              producer.beginTransaction();
              producer.send(record).get();
              producer.commitTransaction();
          }
      }
  
      private void assertDurationAtLeast(KafkaProducer<?, ?> producer, String name, double floor) {
          getAndAssertDurationAtLeast(producer, name, floor);
      }
  
      private double getAndAssertDurationAtLeast(KafkaProducer<?, ?> producer, String name, double floor) {
          double value = getMetricValue(producer, name);
          assertTrue(value >= floor);
          return value;
      }
  
      @Test
      public void testMeasureTransactionDurations() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          Duration tick = Duration.ofSeconds(1);
          Time time = new MockTime(tick.toMillis());
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
          try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
              new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
              assertDurationAtLeast(producer, "txn-init-time-ns-total", tick.toNanos());
  
              client.prepareResponse(addOffsetsToTxnResponse(Errors.NONE));
              client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
              client.prepareResponse(txnOffsetsCommitResponse(Collections.singletonMap(
                  new TopicPartition("topic", 0), Errors.NONE)));
              client.prepareResponse(endTxnResponse(Errors.NONE));
              producer.beginTransaction();
              double beginFirst = getAndAssertDurationAtLeast(producer, "txn-begin-time-ns-total", tick.toNanos());
              producer.sendOffsetsToTransaction(Collections.singletonMap(
                      new TopicPartition("topic", 0),
                      new OffsetAndMetadata(5L)),
                      new ConsumerGroupMetadata("group"));
              double sendOffFirst = getAndAssertDurationAtLeast(producer, "txn-send-offsets-time-ns-total", tick.toNanos());
              producer.commitTransaction();
              double commitFirst = getAndAssertDurationAtLeast(producer, "txn-commit-time-ns-total", tick.toNanos());
  
              client.prepareResponse(addOffsetsToTxnResponse(Errors.NONE));
              client.prepareResponse(txnOffsetsCommitResponse(Collections.singletonMap(
                  new TopicPartition("topic", 0), Errors.NONE)));
              client.prepareResponse(endTxnResponse(Errors.NONE));
              producer.beginTransaction();
              assertDurationAtLeast(producer, "txn-begin-time-ns-total", beginFirst + tick.toNanos());
              producer.sendOffsetsToTransaction(Collections.singletonMap(
                      new TopicPartition("topic", 0),
                      new OffsetAndMetadata(10L)),
                      new ConsumerGroupMetadata("group"));
              assertDurationAtLeast(producer, "txn-send-offsets-time-ns-total", sendOffFirst + tick.toNanos());
              producer.commitTransaction();
              assertDurationAtLeast(producer, "txn-commit-time-ns-total", commitFirst + tick.toNanos());
          }
      }
  
      @Test
      public void testSendTxnOffsetsWithGroupMetadata() {
          final short maxVersion = (short) 3;
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
          client.setNodeApiVersions(NodeApiVersions.create(ApiKeys.TXN_OFFSET_COMMIT.id, (short) 0, maxVersion));
  
          Node node = metadata.fetch().nodes().get(0);
          client.throttle(node, 5000);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
          client.prepareResponse(addOffsetsToTxnResponse(Errors.NONE));
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          String groupId = "group";
          String memberId = "member";
          int generationId = 5;
          String groupInstanceId = "instance";
          client.prepareResponse(request -> {
              TxnOffsetCommitRequestData data = ((TxnOffsetCommitRequest) request).data();
              return data.groupId().equals(groupId) &&
                  data.memberId().equals(memberId) &&
                  data.generationId() == generationId &&
                  data.groupInstanceId().equals(groupInstanceId);
          }, txnOffsetsCommitResponse(Collections.singletonMap(
              new TopicPartition("topic", 0), Errors.NONE)));
          client.prepareResponse(endTxnResponse(Errors.NONE));
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                                                                 new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
              producer.beginTransaction();
              ConsumerGroupMetadata groupMetadata = new ConsumerGroupMetadata(groupId,
                  generationId, memberId, Optional.of(groupInstanceId));
  
              producer.sendOffsetsToTransaction(Collections.emptyMap(), groupMetadata);
              producer.commitTransaction();
          }
      }
  
      @Test
      public void testNullGroupMetadataInSendOffsets() {
          verifyInvalidGroupMetadata(null);
      }
  
      @Test
      public void testInvalidGenerationIdAndMemberIdCombinedInSendOffsets() {
          verifyInvalidGroupMetadata(new ConsumerGroupMetadata("group", 2, JoinGroupRequest.UNKNOWN_MEMBER_ID, Optional.empty()));
      }
  
      @Test
      public void testClientInstanceId() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
  
          ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
          clientTelemetryReporter.configure(any());
  
          try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
              mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
  
              ClientTelemetrySender clientTelemetrySender = mock(ClientTelemetrySender.class);
              Uuid expectedUuid = Uuid.randomUuid();
              when(clientTelemetryReporter.telemetrySender()).thenReturn(clientTelemetrySender);
              when(clientTelemetrySender.clientInstanceId(any())).thenReturn(Optional.of(expectedUuid));
  
              try (KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer())) {
                  Uuid uuid = producer.clientInstanceId(Duration.ofMillis(0));
                  assertEquals(expectedUuid, uuid);
              }
          }
      }
  
      @Test
      public void testClientInstanceIdInvalidTimeout() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
  
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer())) {
              Exception exception = assertThrows(IllegalArgumentException.class, () -> producer.clientInstanceId(Duration.ofMillis(-1)));
              assertEquals("The timeout cannot be negative.", exception.getMessage());
          }
      }
  
      @Test
      public void testClientInstanceIdNoTelemetryReporterRegistered() {
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.setProperty(ProducerConfig.ENABLE_METRICS_PUSH_CONFIG, "false");
  
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer())) {
              Exception exception = assertThrows(IllegalStateException.class, () -> producer.clientInstanceId(Duration.ofMillis(0)));
              assertEquals("Telemetry is not enabled. Set config `enable.metrics.push` to `true`.", exception.getMessage());
          }
      }
  
      private void verifyInvalidGroupMetadata(ConsumerGroupMetadata groupMetadata) {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          Node node = metadata.fetch().nodes().get(0);
          client.throttle(node, 5000);
  
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE));
          client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE));
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
              new StringSerializer(), metadata, client, null, time)) {
              producer.initTransactions();
              producer.beginTransaction();
              assertThrows(IllegalArgumentException.class,
                  () -> producer.sendOffsetsToTransaction(Collections.emptyMap(), groupMetadata));
          }
      }
  
      private InitProducerIdResponse initProducerIdResponse(long producerId, short producerEpoch, Errors error) {
          InitProducerIdResponseData responseData = new InitProducerIdResponseData()
                  .setErrorCode(error.code())
                  .setProducerEpoch(producerEpoch)
                  .setProducerId(producerId)
                  .setThrottleTimeMs(0);
          return new InitProducerIdResponse(responseData);
      }
  
      private AddOffsetsToTxnResponse addOffsetsToTxnResponse(Errors error) {
          return new AddOffsetsToTxnResponse(new AddOffsetsToTxnResponseData()
                                                 .setErrorCode(error.code())
                                                 .setThrottleTimeMs(10));
      }
  
      private TxnOffsetCommitResponse txnOffsetsCommitResponse(Map<TopicPartition, Errors> errorMap) {
          return new TxnOffsetCommitResponse(10, errorMap);
      }
  
      private EndTxnResponse endTxnResponse(Errors error) {
          return new EndTxnResponse(new EndTxnResponseData()
                                        .setErrorCode(error.code())
                                        .setThrottleTimeMs(0));
      }
  
      @Test
      public void testOnlyCanExecuteCloseAfterInitTransactionsTimeout() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
  
          Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                  metadata, client, null, time);
          assertThrows(TimeoutException.class, producer::initTransactions);
          // other transactional operations should not be allowed if we catch the error after initTransactions failed
          try {
              assertThrows(IllegalStateException.class, producer::beginTransaction);
          } finally {
              producer.close(Duration.ofMillis(0));
          }
      }
  
      @Test
      public void testSendToInvalidTopic() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "15000");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, emptyMap());
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
  
          Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                  metadata, client, null, time);
  
          String invalidTopicName = "topic abc"; // Invalid topic name due to space
          ProducerRecord<String, String> record = new ProducerRecord<>(invalidTopicName, "HelloKafka");
  
          List<MetadataResponse.TopicMetadata> topicMetadata = new ArrayList<>();
          topicMetadata.add(new MetadataResponse.TopicMetadata(Errors.INVALID_TOPIC_EXCEPTION,
                  invalidTopicName, false, Collections.emptyList()));
          MetadataResponse updateResponse =  RequestTestUtils.metadataResponse(
                  new ArrayList<>(initialUpdateResponse.brokers()),
                  initialUpdateResponse.clusterId(),
                  initialUpdateResponse.controller().id(),
                  topicMetadata);
          client.prepareMetadataUpdate(updateResponse);
  
          Future<RecordMetadata> future = producer.send(record);
  
          assertEquals(Collections.singleton(invalidTopicName),
                  metadata.fetch().invalidTopics(), "Cluster has incorrect invalid topic list.");
          TestUtils.assertFutureThrows(InvalidTopicException.class, future);
  
          producer.close(Duration.ofMillis(0));
      }
  
      @Test
      public void testCloseWhenWaitingForMetadataUpdate() throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.MAX_VALUE);
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
  
          // Simulate a case where metadata for a particular topic is not available. This will cause KafkaProducer#send to
          // block in Metadata#awaitUpdate for the configured max.block.ms. When close() is invoked, KafkaProducer#send should
          // return with a KafkaException.
          String topicName = "test";
          Time time = Time.SYSTEM;
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, emptyMap());
          ProducerMetadata metadata = new ProducerMetadata(0, 0, Long.MAX_VALUE, Long.MAX_VALUE,
                  new LogContext(), new ClusterResourceListeners(), time);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
          MockClient client = new MockClient(time, metadata);
  
          Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                  metadata, client, null, time);
  
          ExecutorService executor = Executors.newSingleThreadExecutor();
          final AtomicReference<Exception> sendException = new AtomicReference<>();
  
          try {
              executor.submit(() -> {
                  try {
                      // Metadata for topic "test" will not be available which will cause us to block indefinitely until
                      // KafkaProducer#close is invoked.
                      producer.send(new ProducerRecord<>(topicName, "key", "value"));
                      fail();
                  } catch (Exception e) {
                      sendException.set(e);
                  }
              });
  
              // Wait until metadata update for the topic has been requested
              TestUtils.waitForCondition(() -> metadata.containsTopic(topicName),
                      "Timeout when waiting for topic to be added to metadata");
              producer.close(Duration.ofMillis(0));
              TestUtils.waitForCondition(() -> sendException.get() != null, "No producer exception within timeout");
              assertEquals(KafkaException.class, sendException.get().getClass());
          } finally {
              executor.shutdownNow();
          }
      }
  
      @Test
      public void testTransactionalMethodThrowsWhenSenderClosed() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "this-is-a-transactional-id");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, emptyMap());
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
  
          Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                  metadata, client, null, time);
          producer.close();
          assertThrows(IllegalStateException.class, producer::initTransactions);
      }
  
      @Test
      public void testCloseIsForcedOnPendingFindCoordinator() throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "this-is-a-transactional-id");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("testTopic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
  
          Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                  metadata, client, null, time);
  
          ExecutorService executorService = Executors.newSingleThreadExecutor();
          CountDownLatch assertionDoneLatch = new CountDownLatch(1);
          executorService.submit(() -> {
              assertThrows(KafkaException.class, producer::initTransactions);
              assertionDoneLatch.countDown();
          });
  
          client.waitForRequests(1, 2000);
          producer.close(Duration.ofMillis(1000));
          assertionDoneLatch.await(5000, TimeUnit.MILLISECONDS);
      }
  
      @Test
      public void testCloseIsForcedOnPendingInitProducerId() throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "this-is-a-transactional-id");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("testTopic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
  
          Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                  metadata, client, null, time);
  
          ExecutorService executorService = Executors.newSingleThreadExecutor();
          CountDownLatch assertionDoneLatch = new CountDownLatch(1);
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "this-is-a-transactional-id", NODE));
          executorService.submit(() -> {
              assertThrows(KafkaException.class, producer::initTransactions);
              assertionDoneLatch.countDown();
          });
  
          client.waitForRequests(1, 2000);
          producer.close(Duration.ofMillis(1000));
          assertionDoneLatch.await(5000, TimeUnit.MILLISECONDS);
      }
  
      @Test
      public void testCloseIsForcedOnPendingAddOffsetRequest() throws InterruptedException {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "this-is-a-transactional-id");
  
          Time time = new MockTime();
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("testTopic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
          metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
  
          MockClient client = new MockClient(time, metadata);
  
          Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                  metadata, client, null, time);
  
          ExecutorService executorService = Executors.newSingleThreadExecutor();
          CountDownLatch assertionDoneLatch = new CountDownLatch(1);
          client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "this-is-a-transactional-id", NODE));
          executorService.submit(() -> {
              assertThrows(KafkaException.class, producer::initTransactions);
              assertionDoneLatch.countDown();
          });
  
          client.waitForRequests(1, 2000);
          producer.close(Duration.ofMillis(1000));
          assertionDoneLatch.await(5000, TimeUnit.MILLISECONDS);
      }
  
      @Test
      public void testProducerJmxPrefix() throws  Exception {
          Map<String, Object> props = new HashMap<>();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.put("client.id", "client-1");
  
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                  props, new StringSerializer(), new StringSerializer())) {
              MBeanServer server = ManagementFactory.getPlatformMBeanServer();
              MetricName testMetricName = producer.metrics.metricName("test-metric",
                      "grp1", "test metric");
              producer.metrics.addMetric(testMetricName, new Avg());
              assertNotNull(server.getObjectInstance(new ObjectName("kafka.producer:type=grp1,client-id=client-1")));
          }
      }
  
      private static ProducerMetadata newMetadata(long refreshBackoffMs, long refreshBackoffMaxMs, long expirationMs) {
          return new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, expirationMs, DEFAULT_METADATA_IDLE_MS,
                  new LogContext(), new ClusterResourceListeners(), Time.SYSTEM);
      }
  
      @Test
      public void configurableObjectsShouldSeeGeneratedClientId() {
          Properties props = new Properties();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, SerializerForClientId.class.getName());
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, SerializerForClientId.class.getName());
          props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, PartitionerForClientId.class.getName());
          props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorForClientId.class.getName());
  
          KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props);
          assertNotNull(producer.getClientId());
          assertNotEquals(0, producer.getClientId().length());
          assertEquals(4, CLIENT_IDS.size());
          CLIENT_IDS.forEach(id -> assertEquals(id, producer.getClientId()));
          producer.close();
      }
  
      @Test
      public void testUnusedConfigs() {
          Map<String, Object> props = new HashMap<>();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          props.put(SslConfigs.SSL_PROTOCOL_CONFIG, "TLS");
          ProducerConfig config = new ProducerConfig(ProducerConfig.appendSerializerToConfig(props,
                  new StringSerializer(), new StringSerializer()));
  
          assertTrue(config.unused().contains(SslConfigs.SSL_PROTOCOL_CONFIG));
  
          try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(config, null, null,
                  null, null, null, null, Time.SYSTEM)) {
              assertTrue(config.unused().contains(SslConfigs.SSL_PROTOCOL_CONFIG));
          }
      }
  
      @Test
      public void testNullTopicName() {
          // send a record with null topic should fail
          assertThrows(IllegalArgumentException.class, () -> new ProducerRecord<>(null, 1,
              "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8)));
      }
  
      @Test
      public void testCallbackAndInterceptorHandleError() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "1000");
          configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MockProducerInterceptor.class.getName());
          configs.put(MockProducerInterceptor.APPEND_STRING_PROP, "something");
  
  
          Time time = new MockTime();
          ProducerMetadata producerMetadata = newMetadata(0, 0, Long.MAX_VALUE);
          MockClient client = new MockClient(time, producerMetadata);
  
          String invalidTopicName = "topic abc"; // Invalid topic name due to space
  
          ProducerInterceptors<String, String> producerInterceptors =
                  new ProducerInterceptors<>(Collections.singletonList(new MockProducerInterceptor()), null);
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                  producerMetadata, client, producerInterceptors, time)) {
              ProducerRecord<String, String> record = new ProducerRecord<>(invalidTopicName, "HelloKafka");
  
              // Here's the important piece of the test. Let's make sure that the RecordMetadata we get
              // is non-null and adheres to the onCompletion contract.
              Callback callBack = (recordMetadata, exception) -> {
                  assertNotNull(exception);
                  assertNotNull(recordMetadata);
  
                  assertNotNull(recordMetadata.topic(), "Topic name should be valid even on send failure");
                  assertEquals(invalidTopicName, recordMetadata.topic());
  
                  assertFalse(recordMetadata.hasOffset());
                  assertEquals(ProduceResponse.INVALID_OFFSET, recordMetadata.offset());
  
                  assertFalse(recordMetadata.hasTimestamp());
                  assertEquals(RecordBatch.NO_TIMESTAMP, recordMetadata.timestamp());
  
                  assertEquals(-1, recordMetadata.serializedKeySize());
                  assertEquals(-1, recordMetadata.serializedValueSize());
                  assertEquals(-1, recordMetadata.partition());
              };
  
              producer.send(record, callBack);
              assertEquals(1, MockProducerInterceptor.ON_ACKNOWLEDGEMENT_COUNT.intValue());
          }
      }
  
      @Test
      public void shouldNotInvokeFlushInCallback() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          // only test in idempotence disabled producer for simplicity
          configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
          AtomicReference<KafkaException> kafkaException = new AtomicReference<>();
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
              new StringSerializer(), metadata, client, null, time)) {
              producer.send(
                  new ProducerRecord<>("topic", "value"),
                  (recordMetadata, exception) -> kafkaException.set(assertThrows(KafkaException.class, producer::flush))
              );
          }
  
          assertNotNull(kafkaException.get());
          assertEquals("KafkaProducer.flush() invocation inside a callback is not permitted because it may lead to deadlock.",
              kafkaException.get().getMessage());
      }
  
      @Test
      public void negativePartitionShouldThrow() {
          Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
          configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, BuggyPartitioner.class.getName());
  
          Time time = new MockTime(1);
          MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
          ProducerMetadata metadata = newMetadata(0, 0, Long.MAX_VALUE);
  
          MockClient client = new MockClient(time, metadata);
          client.updateMetadata(initialUpdateResponse);
  
          try (Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                  new StringSerializer(), metadata, client, null, time)) {
              assertThrows(IllegalArgumentException.class, () -> producer.send(new ProducerRecord<>("topic", "key", "value")));
          }
      }
  
      @Test
      public void testPartitionAddedToTransaction() throws Exception {
          StringSerializer serializer = new StringSerializer();
          KafkaProducerTestContext<String> ctx = new KafkaProducerTestContext<>(testInfo, serializer);
  
          String topic = "foo";
          TopicPartition topicPartition = new TopicPartition(topic, 0);
          Cluster cluster = TestUtils.singletonCluster(topic, 1);
  
          when(ctx.sender.isRunning()).thenReturn(true);
          when(ctx.metadata.fetch()).thenReturn(cluster);
  
          long timestamp = ctx.time.milliseconds();
          ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, timestamp, "key", "value");
          FutureRecordMetadata future = expectAppend(ctx, record, topicPartition, cluster);
  
          try (KafkaProducer<String, String> producer = ctx.newKafkaProducer()) {
              assertEquals(future, producer.send(record));
              assertFalse(future.isDone());
              verify(ctx.transactionManager).maybeAddPartition(topicPartition);
          }
      }
  
      private <T> FutureRecordMetadata expectAppend(
          KafkaProducerTestContext<T> ctx,
          ProducerRecord<T, T> record,
          TopicPartition initialSelectedPartition,
          Cluster cluster
      ) throws InterruptedException {
          byte[] serializedKey = ctx.serializer.serialize(topic, record.key());
          byte[] serializedValue = ctx.serializer.serialize(topic, record.value());
          long timestamp = record.timestamp() == null ? ctx.time.milliseconds() : record.timestamp();
  
          ProduceRequestResult requestResult = new ProduceRequestResult(initialSelectedPartition);
          FutureRecordMetadata futureRecordMetadata = new FutureRecordMetadata(
              requestResult,
              5,
              timestamp,
              serializedKey.length,
              serializedValue.length,
              ctx.time
          );
  
          when(ctx.partitioner.partition(
              initialSelectedPartition.topic(),
              record.key(),
              serializedKey,
              record.value(),
              serializedValue,
              cluster
          )).thenReturn(initialSelectedPartition.partition());
  
          when(ctx.accumulator.append(
              eq(initialSelectedPartition.topic()),            // 0
              eq(initialSelectedPartition.partition()),        // 1
              eq(timestamp),                                   // 2
              eq(serializedKey),                               // 3
              eq(serializedValue),                             // 4
              eq(Record.EMPTY_HEADERS),                        // 5
              any(RecordAccumulator.AppendCallbacks.class),    // 6 <--
              anyLong(),
              anyLong(),
              any()
          )).thenAnswer(invocation -> {
              RecordAccumulator.AppendCallbacks callbacks =
                  (RecordAccumulator.AppendCallbacks) invocation.getArguments()[6];
              callbacks.setPartition(initialSelectedPartition.partition());
              return new RecordAccumulator.RecordAppendResult(
                  futureRecordMetadata,
                  false,
                  false,
                  0);
          });
  
          return futureRecordMetadata;
      }
  
      private static final List<String> CLIENT_IDS = new ArrayList<>();
  
      public static class SerializerForClientId implements Serializer<byte[]> {
          @Override
          public void configure(Map<String, ?> configs, boolean isKey) {
              CLIENT_IDS.add(configs.get(ProducerConfig.CLIENT_ID_CONFIG).toString());
          }
  
          @Override
          public byte[] serialize(String topic, byte[] data) {
              return data;
          }
      }
  
      public static class PartitionerForClientId implements Partitioner {
  
          @Override
          public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
              return 0;
          }
  
          @Override
          public void close() {
  
          }
  
          @Override
          public void configure(Map<String, ?> configs) {
              CLIENT_IDS.add(configs.get(ProducerConfig.CLIENT_ID_CONFIG).toString());
          }
      }
  
      public static class ProducerInterceptorForHeaders implements ProducerInterceptor<byte[], byte[]> {
  
          @Override
          public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> record) {
              return record;
          }
  
          @Override
          public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
              RecordHeaders recordHeaders = (RecordHeaders) headers;
              // Ensure that the headers are read-only, no matter send success or send failure
              assertTrue(recordHeaders.isReadOnly());
          }
  
          @Override
          public void close() {
          }
  
          @Override
          public void configure(Map<String, ?> configs) {
          }
      }
  
      public static class ProducerInterceptorForClientId implements ProducerInterceptor<byte[], byte[]> {
  
          @Override
          public ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> record) {
              return record;
          }
  
          @Override
          public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
          }
  
          @Override
          public void close() {
          }
  
          @Override
          public void configure(Map<String, ?> configs) {
              CLIENT_IDS.add(configs.get(ProducerConfig.CLIENT_ID_CONFIG).toString());
          }
      }
  
      public static class BuggyPartitioner implements Partitioner {
  
          @Override
          public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
              return -1;
          }
  
          @Override
          public void close() {
          }
  
          @Override
          public void configure(Map<String, ?> configs) {
          }
      }
  
      private static class KafkaProducerTestContext<T> {
          private final TestInfo testInfo;
          private final Map<String, Object> configs;
          private final Serializer<T> serializer;
          private final Partitioner partitioner = mock(Partitioner.class);
          private final Sender.SenderThread senderThread = mock(Sender.SenderThread.class);
          private final List<ProducerInterceptor<T, T>> interceptors = new ArrayList<>();
          private ProducerMetadata metadata = mock(ProducerMetadata.class);
          private RecordAccumulator accumulator = mock(RecordAccumulator.class);
          private Sender sender = mock(Sender.class);
          private TransactionManager transactionManager = mock(TransactionManager.class);
          private Time time = new MockTime();
          private final Metrics metrics = new Metrics(time);
  
          public KafkaProducerTestContext(
              TestInfo testInfo,
              Serializer<T> serializer
          ) {
              this(testInfo, new HashMap<>(), serializer);
          }
  
          public KafkaProducerTestContext(
              TestInfo testInfo,
              Map<String, Object> configs,
              Serializer<T> serializer
          ) {
              this.testInfo = testInfo;
              this.configs = configs;
              this.serializer = serializer;
  
              if (!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
                  configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              }
          }
  
          public KafkaProducerTestContext<T> setProducerMetadata(ProducerMetadata metadata) {
              this.metadata = metadata;
              return this;
          }
  
          public KafkaProducerTestContext<T> setAccumulator(RecordAccumulator accumulator) {
              this.accumulator = accumulator;
              return this;
          }
  
          public KafkaProducerTestContext<T> setSender(Sender sender) {
              this.sender = sender;
              return this;
          }
  
          public KafkaProducerTestContext<T> setTransactionManager(TransactionManager transactionManager) {
              this.transactionManager = transactionManager;
              return this;
          }
  
          public KafkaProducerTestContext<T> addInterceptor(ProducerInterceptor<T, T> interceptor) {
              this.interceptors.add(interceptor);
              return this;
          }
  
          public KafkaProducerTestContext<T> setTime(Time time) {
              this.time = time;
              return this;
          }
  
          public KafkaProducer<T, T> newKafkaProducer() {
              LogContext logContext = new LogContext("[Producer test=" + testInfo.getDisplayName() + "] ");
  
              ProducerConfig producerConfig = new ProducerConfig(
                  ProducerConfig.appendSerializerToConfig(configs, serializer, serializer));
  
              ProducerInterceptors<T, T> interceptors = new ProducerInterceptors<>(this.interceptors, metrics);
  
              return new KafkaProducer<>(
                  producerConfig,
                  logContext,
                  metrics,
                  serializer,
                  serializer,
                  metadata,
                  accumulator,
                  transactionManager,
                  sender,
                  interceptors,
                  partitioner,
                  time,
                  senderThread,
                  Optional.empty()
              );
          }
      }
  
      @Test
      void testDeliveryTimeoutAndLingerMsConfig() {
          final Map<String, Object> configs = new HashMap<>();
          configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testDeliveryTimeoutAndLingerMsConfig");
          configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
          configs.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 1000);
          configs.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
          configs.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1);
  
          // delivery.timeout.ms should be equal to or larger than linger.ms + request.timeout.ms
          assertThrows(KafkaException.class, () -> new KafkaProducer<>(configs, new StringSerializer(), new StringSerializer()));
  
          configs.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 1000);
          configs.put(ProducerConfig.LINGER_MS_CONFIG, 999);
          configs.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1);
  
          assertDoesNotThrow(() -> new KafkaProducer<>(configs, new StringSerializer(), new StringSerializer()).close());
      }
  
      @SuppressWarnings("deprecation")
      private ProduceResponse produceResponse(TopicIdPartition topicIdPartition, long offset, Errors error, int throttleTimeMs, int logStartOffset) {
          ProduceResponse.PartitionResponse resp = new ProduceResponse.PartitionResponse(error, offset, RecordBatch.NO_TIMESTAMP, logStartOffset);
          Map<TopicIdPartition, ProduceResponse.PartitionResponse> partResp = singletonMap(topicIdPartition, resp);
          return new ProduceResponse(partResp, throttleTimeMs);
      }
  
      @Test
      public void testSubscribingCustomMetricsDoesntAffectProducerMetrics() {
          Map<String, Object> props = new HashMap<>();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
  
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(
              props, new StringSerializer(), new StringSerializer())) {
              Map<MetricName, KafkaMetric> customMetrics = customMetrics();
              customMetrics.forEach((name, metric) -> producer.registerMetricForSubscription(metric));
  
              Map<MetricName, ? extends Metric> producerMetrics = producer.metrics();
              customMetrics.forEach((name, metric) -> assertFalse(producerMetrics.containsKey(name)));
          }
      }
  
      @Test
      public void testUnSubscribingNonExisingMetricsDoesntCauseError() {
          Map<String, Object> props = new HashMap<>();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
  
          try (KafkaProducer<String, String> producer = new KafkaProducer<>(
              props, new StringSerializer(), new StringSerializer())) {
              Map<MetricName, KafkaMetric> customMetrics = customMetrics();
              //Metrics never registered but removed should not cause an error
              customMetrics.forEach((name, metric) -> assertDoesNotThrow(() -> producer.unregisterMetricFromSubscription(metric)));
          }
      }
  
      @Test
      public void testSubscribingCustomMetricsWithSameNameDoesntAffectProducerMetrics() {
          try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
              appender.setClassLogger(KafkaProducer.class, Level.DEBUG);
              Map<String, Object> props = new HashMap<>();
              props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                      props, new StringSerializer(), new StringSerializer())) {
                  KafkaMetric existingMetricToAdd = (KafkaMetric) producer.metrics().entrySet().iterator().next().getValue();
                  producer.registerMetricForSubscription(existingMetricToAdd);
                  final String expectedMessage = String.format("Skipping registration for metric %s. Existing producer metrics cannot be overwritten.", existingMetricToAdd.metricName());
                  assertTrue(appender.getMessages().stream().anyMatch(m -> m.contains(expectedMessage)));
              }
          }
      }
  
      @Test
      public void testUnsubscribingCustomMetricWithSameNameAsExistingMetricDoesntAffectProducerMetric() {
          try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
              appender.setClassLogger(KafkaProducer.class, Level.DEBUG);
              Map<String, Object> props = new HashMap<>();
              props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                      props, new StringSerializer(), new StringSerializer())) {
                  KafkaMetric existingMetricToRemove = (KafkaMetric) producer.metrics().entrySet().iterator().next().getValue();
                  producer.unregisterMetricFromSubscription(existingMetricToRemove);
                  final String expectedMessage = String.format("Skipping unregistration for metric %s. Existing producer metrics cannot be removed.", existingMetricToRemove.metricName());
                  assertTrue(appender.getMessages().stream().anyMatch(m -> m.contains(expectedMessage)));
              }
          }
      }
  
      @Test
      public void testShouldOnlyCallMetricReporterMetricChangeOnceWithExistingProducerMetric() {
          try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
              ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
              clientTelemetryReporter.configure(any());
              mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
  
              Map<String, Object> props = new HashMap<>();
              props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                      props, new StringSerializer(), new StringSerializer())) {
                  KafkaMetric existingMetric = (KafkaMetric) producer.metrics().entrySet().iterator().next().getValue();
                  producer.registerMetricForSubscription(existingMetric);
                  // This test would fail without the check as the existing metric is registered in the producer on startup
                  Mockito.verify(clientTelemetryReporter, atMostOnce()).metricChange(existingMetric);
              }
          }
      }
  
      @Test
      public void testShouldNotCallMetricReporterMetricRemovalWithExistingProducerMetric() {
          try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
              ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
              clientTelemetryReporter.configure(any());
              mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
  
              Map<String, Object> props = new HashMap<>();
              props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                      props, new StringSerializer(), new StringSerializer())) {
                  KafkaMetric existingMetric = (KafkaMetric) producer.metrics().entrySet().iterator().next().getValue();
                  producer.unregisterMetricFromSubscription(existingMetric);
                  // This test would fail without the check as the existing metric is registered in the consumer on startup
                  Mockito.verify(clientTelemetryReporter, never()).metricRemoval(existingMetric);
              }
          }
      }
  
  
      private Map<MetricName, KafkaMetric> customMetrics() {
          MetricConfig metricConfig = new MetricConfig();
          Object lock = new Object();
          MetricName metricNameOne = new MetricName("metricOne", "stream-metrics", "description for metric one", new HashMap<>());
          MetricName metricNameTwo = new MetricName("metricTwo", "stream-metrics", "description for metric two", new HashMap<>());
  
          KafkaMetric streamClientMetricOne = new KafkaMetric(lock, metricNameOne, (Measurable) (m, now) -> 1.0, metricConfig, Time.SYSTEM);
          KafkaMetric streamClientMetricTwo = new KafkaMetric(lock, metricNameTwo, (Measurable) (m, now) -> 2.0, metricConfig, Time.SYSTEM);
          return Map.of(metricNameOne, streamClientMetricOne, metricNameTwo, streamClientMetricTwo);
      }
  
      @Test
      void testMonitorablePlugins() {
          try {
              String clientId = "testMonitorablePlugins";
              Map<String, Object> configs = new HashMap<>();
              configs.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
              configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
              configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, MonitorableSerializer.class.getName());
              configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, MonitorableSerializer.class.getName());
              configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MonitorablePartitioner.class.getName());
              configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MonitorableInterceptor.class.getName());
              configs.put(MockProducerInterceptor.APPEND_STRING_PROP, "");
  
              KafkaProducer<String, String> producer = new KafkaProducer<>(configs);
              Map<MetricName, ? extends Metric> metrics = producer.metrics();
  
              MetricName expectedKeySerializerMetric = expectedMetricName(
                      clientId,
                      ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                      MonitorableSerializer.class);
              assertTrue(metrics.containsKey(expectedKeySerializerMetric));
              assertEquals(VALUE, metrics.get(expectedKeySerializerMetric).metricValue());
  
              MetricName expectedValueSerializerMetric = expectedMetricName(
                      clientId,
                      ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                      MonitorableSerializer.class);
              assertTrue(metrics.containsKey(expectedValueSerializerMetric));
              assertEquals(VALUE, metrics.get(expectedValueSerializerMetric).metricValue());
  
              MetricName expectedPartitionerMetric = expectedMetricName(
                      clientId,
                      ProducerConfig.PARTITIONER_CLASS_CONFIG,
                      MonitorablePartitioner.class);
              assertTrue(metrics.containsKey(expectedPartitionerMetric));
              assertEquals(VALUE, metrics.get(expectedPartitionerMetric).metricValue());
  
              MetricName expectedInterceptorMetric = expectedMetricName(
                      clientId,
                      ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
                      MonitorableInterceptor.class);
              assertTrue(metrics.containsKey(expectedInterceptorMetric));
              assertEquals(VALUE, metrics.get(expectedInterceptorMetric).metricValue());
  
              producer.close();
              metrics = producer.metrics();
              assertFalse(metrics.containsKey(expectedKeySerializerMetric));
              assertFalse(metrics.containsKey(expectedValueSerializerMetric));
              assertFalse(metrics.containsKey(expectedPartitionerMetric));
              assertFalse(metrics.containsKey(expectedInterceptorMetric));
          } finally {
              MockProducerInterceptor.resetCounters();
          }
      }
  
      private MetricName expectedMetricName(String clientId, String config, Class<?> clazz) {
          Map<String, String> expectedTags = new LinkedHashMap<>();
          expectedTags.put("client-id", clientId);
          expectedTags.put("config", config);
          expectedTags.put("class", clazz.getSimpleName());
          expectedTags.putAll(TAGS);
          return new MetricName(NAME, "plugins", DESCRIPTION, expectedTags);
      }
  
      private static final String NAME = "name";
      private static final String DESCRIPTION = "description";
      private static final LinkedHashMap<String, String> TAGS = new LinkedHashMap<>();
      private static final double VALUE = 123.0;
  
      static {
          TAGS.put("t1", "v1");
      }
  
      public static class MonitorableSerializer extends MockSerializer implements Monitorable {
  
          @Override
          public void withPluginMetrics(PluginMetrics metrics) {
              MetricName name = metrics.metricName(NAME, DESCRIPTION, TAGS);
              metrics.addMetric(name, (Measurable) (config, now) -> VALUE);
          }
      }
  
      public static class MonitorablePartitioner extends MockPartitioner implements Monitorable {
  
          @Override
          public void withPluginMetrics(PluginMetrics metrics) {
              MetricName name = metrics.metricName(NAME, DESCRIPTION, TAGS);
              metrics.addMetric(name, (Measurable) (config, now) -> VALUE);
          }
      }
  
      public static class MonitorableInterceptor extends MockProducerInterceptor implements Monitorable {
  
          @Override
          public void withPluginMetrics(PluginMetrics metrics) {
              MetricName name = metrics.metricName(NAME, DESCRIPTION, TAGS);
              metrics.addMetric(name, (Measurable) (config, now) -> VALUE);
          }
      }
  }
  