mongoDB & Spark: "com.mongodb.MongoSocketReadException: Prematurely reached end of stream"

user_3352382 0
Viewed 666

I have a Java application that processes a Kafka stream of avro messages and for each message performs a query on a mongoDB collection.

After a few dozens messages being properly processed, the application stops running and throws "com.mongodb.MongoSocketReadException: Prematurely reached end of stream".

Here's the code:

    JavaPairInputDStream<String, byte[]> directKafkaStream = KafkaUtils.createDirectStream(jsc,
            String.class, byte[].class, StringDecoder.class, DefaultDecoder.class, kafkaParams, topics);

    directKafkaStream.foreachRDD(rdd ->{

        rdd.foreach(avroRecord -> {

            byte[] encodedAvroData = avroRecord._2;
            LocationType t = deserialize(encodedAvroData);

            MongoClientOptions.Builder options_builder = new MongoClientOptions.Builder();
            options_builder.maxConnectionIdleTime(60000);
            MongoClientOptions options = options_builder.build();
            MongoClient mongo = new MongoClient ("localhost:27017", options);

            MongoDatabase database = mongo.getDatabase("DB");
            MongoCollection<Document> collection = database.getCollection("collection");

            Document myDoc = collection.find(eq("key", 4)).first();
            System.out.println(myDoc);

        });
    });
apache-kafka apache-spark mongodb java
1 Answers

First you should not open a mongo connection for each record ! Then you should close your mongo connection.

Mongo doesn't like when you open many (hundreds, thousands ?) without closing them.

Here is an exemple of what you can do to open mongo connection over a RDD :

directKafkaStream.foreachRDD(rdd ->{
    rdd.foreachPartition(it -> {

        // Opens only 1 connection per partition
        MongoClient mongo = new MongoClient ("localhost:27017");
        MongoDatabase database = mongo.getDatabase("DB");
        MongoCollection<Document> collection = database.getCollection("collection");

        while (it.hasNext()) {
            byte[] encodedAvroData = it.next()._2;
            LocationType t = deserialize(encodedAvroData);

            Document myDoc = collection.find(eq("key", 4)).first();
            System.out.println(myDoc);
        }

        mongo.close();
    });
});
user_55206460
Related