@Override
    public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
        return send(record, null);
    }
@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);
        }
    }