非SpringBoot核心Java应用的Kafka、MongoDB及映射类JUnit测试需求

一、映射类测试（纯JUnit断言） #

映射类是纯POJO转换逻辑，无需外部依赖，直接用JUnit断言验证字段映射正确性。

假设你的映射类为JsonToMongoMapper，负责将Kafka的SourceMessage转换为Mongo的TargetDocument：

import org.junit.Test;
import static org.junit.Assert.*;

public class JsonToMongoMapperTest {

    @Test
    public void testNormalFieldMapping() {
        // 构造源消息
        SourceMessage source = new SourceMessage();
        source.setFieldA("demo-value");
        source.setFieldC(100);

        // 执行映射
        TargetDocument target = JsonToMongoMapper.map(source);

        // 断言映射结果符合预期
        assertEquals("demo-value", target.getFieldB());
        assertEquals(100, target.getFieldD());
        assertNotNull(target.getCreatedAt()); // 验证自动生成的字段
    }

    @Test
    public void testNullSourceFieldMapping() {
        SourceMessage source = new SourceMessage();
        source.setFieldA(null);

        TargetDocument target = JsonToMongoMapper.map(source);

        assertNull(target.getFieldB());
    }
}

二、Kafka消费者测试（使用嵌入式Kafka） #

借助第三方嵌入式Kafka库模拟Kafka环境，无需依赖真实集群，测试消息读取与解析逻辑。

Maven依赖 #

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.1</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>io.github.embeddedkafka</groupId>
    <artifactId>embedded-kafka_2.13</artifactId>
    <version>2.8.1</version>
    <scope>test</scope>
</dependency>

测试代码 #

import io.github.embeddedkafka.EmbeddedKafka;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.util.List;
import java.util.Properties;
import static io.github.embeddedkafka.EmbeddedKafka.produce;
import static org.junit.Assert.assertEquals;

public class KafkaMessageConsumerTest {

    private static final String TEST_TOPIC = "test-topic";
    private static KafkaMessageConsumer customConsumer; // 你的自定义消费者类

    @BeforeClass
    public static void initEmbeddedKafka() {
        // 启动嵌入式Kafka
        EmbeddedKafka.start();
        // 初始化自定义消费者，配置指向嵌入式Kafka地址
        Properties consumerProps = new Properties();
        consumerProps.put("bootstrap.servers", EmbeddedKafka.bootstrapServers());
        consumerProps.put("group.id", "test-consumer-group");
        consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        customConsumer = new KafkaMessageConsumer(consumerProps, TEST_TOPIC);
        customConsumer.start();
    }

    @AfterClass
    public static void shutDownEmbeddedKafka() {
        customConsumer.stop();
        EmbeddedKafka.stop();
    }

    @Test
    public void testConsumeAndParseJson() throws InterruptedException {
        // 构造测试JSON消息
        String testJson = "{\"fieldA\":\"kafka-demo\",\"fieldC\":200}";
        // 发送消息到测试Topic
        produce(new ProducerRecord<>(TEST_TOPIC, testJson), EmbeddedKafka.producerConfig());

        // 等待消费者处理消息
        Thread.sleep(1000);
        // 假设消费者提供获取已处理消息的方法
        List<SourceMessage> processedList = customConsumer.getProcessedMessages();

        assertEquals(1, processedList.size());
        assertEquals("kafka-demo", processedList.get(0).getFieldA());
        assertEquals(200, processedList.get(0).getFieldC());
    }
}

三、MongoDB写入测试（使用嵌入式MongoDB） #

用嵌入式MongoDB模拟数据库环境，测试数据写入后的持久化正确性，无需连接真实Mongo实例。

Maven依赖 #

<dependency>
    <groupId>de.flapdoodle.embed</groupId>
    <artifactId>de.flapdoodle.embed.mongo</artifactId>
    <version>3.4.1</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongodb-driver-sync</artifactId>
    <version>4.5.1</version>
    <scope>test</scope>
</dependency>

测试代码 #

import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import de.flapdoodle.embed.mongo.MongodExecutable;
import de.flapdoodle.embed.mongo.MongodStarter;
import de.flapdoodle.embed.mongo.config.MongodConfig;
import de.flapdoodle.embed.mongo.config.Net;
import de.flapdoodle.embed.mongo.distribution.Version;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static com.mongodb.client.model.Filters.eq;
import static org.junit.Assert.assertEquals;

public class MongoWriterTest {

    private MongodExecutable mongodExecutable;
    private MongoWriter customMongoWriter; // 你的自定义Mongo写入类
    private MongoCollection<TargetDocument> testCollection;

    @Before
    public void initEmbeddedMongo() throws Exception {
        // 启动嵌入式MongoDB
        MongodStarter starter = MongodStarter.getDefaultInstance();
        String bindIp = "localhost";
        int port = 27017;
        MongodConfig mongodConfig = MongodConfig.builder()
                .version(Version.Main.PRODUCTION)
                .net(new Net(bindIp, port, false))
                .build();
        mongodExecutable = starter.prepare(mongodConfig);
        mongodExecutable.start();

        // 初始化Mongo客户端与写入类
        MongoDatabase testDb = MongoClients.create("mongodb://" + bindIp + ":" + port).getDatabase("test-db");
        testCollection = testDb.getCollection("test-collection", TargetDocument.class);
        customMongoWriter = new MongoWriter(testDb, "test-collection");
    }

    @After
    public void shutDownEmbeddedMongo() {
        mongodExecutable.stop();
    }

    @Test
    public void testWriteDocumentToMongo() {
        // 构造目标文档
        TargetDocument testDoc = new TargetDocument();
        testDoc.setFieldB("mongo-demo");
        testDoc.setFieldD(300);

        // 执行写入操作
        customMongoWriter.write(testDoc);

        // 查询验证写入结果
        TargetDocument savedDoc = testCollection.find(eq("fieldB", "mongo-demo")).first();
        assertEquals("mongo-demo", savedDoc.getFieldB());
        assertEquals(300, savedDoc.getFieldD());
    }
}

测试思路总结 #

• 映射类：聚焦字段转换逻辑，纯单元测试，仅用JUnit断言即可覆盖场景。
• Kafka消费者：用嵌入式Kafka模拟完整的生产-消费链路，验证消息读取、JSON解析的正确性。
• MongoDB写入：用嵌入式Mongo模拟数据库环境，验证数据持久化后的字段一致性。

内容的提问来源于stack exchange，提问作者Vanaja Jayaraman