Java中的Spark流单元测试

Question

我正在尝试为我的Spark流作业编写单元测试。我的spark streaming作业使用来自MQ的消息，并将其推入Kafka主题。

我的方法是

null

public class StreamingJobTest {

private static KafkaConsumer<String, String> consumer;

@BeforeClass
public static void setUpClass()  {

    Properties properties = new Properties();

    properties.put("bootstrap.servers", "localhost:9090");
    properties.put("subscribe", "topic1");
    properties.put("startingOffsets", "earliest");
    properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    consumer = new KafkaConsumer<String, String>(properties);



}


@Test
public void create_test() {
    String[] arguments = new String[]{};
    ConsumerRecords<String, String> records;

    Thread thread = new Thread(() -> StreamingJob.main(arguments));
    thread.start();

     //send a message to MQ.

    MqSender mqSender = new MqSender();
    mqSender.mqPushMsg("TestMsg");

    //keep polling the kafka topic.

    while(true){
        System.out.println("Polling...");
        records = consumer.poll(100);

        if(!records.isEmpty()){

            thread.interrupt();
            break;
        }

    assertNotNull(records);

    }


}

Answer 1

我自己想通了。我需要在一个单独的行订阅主题。我把它添加到我的属性中。而且groupid在Kafka中是强制性的，我错过了。它现在对我很好。下面是订阅主题的代码。

consumer.subscribe(Arrays.asList("topic1"));