Spring Data Mongodb - How to do Bulk Upsert

spring mongodb spring-boot spring-data-mongodb bulkupdate
Afridi picture Afridi · Nov 16, 2017 · Viewed 9.9k times · Source

I have a list of objects/domains of same Collection that should be inserted if not present in Mongodb database, otherwise it should update an existing records filter by _id.

Although this can be done using Spring data MongoRepositories, but it seem to be:

  1. Too slow!!!(May be it saves records one by one!!! Is this operation executing in Bulk???)
  2. If there is any DuplicateKeyException, it terminates execution(not saving next records, although I want to ignore such exceptions)

Now in case if I use BulkOperations bulkOps = mongoTemplate.bulkOps(BulkMode.UNORDERED, collectionName), it still throws same exception if integrity voilates!!! Although according to Spring doc, using BulkMode.UNORDERED:

Perform bulk operations in parallel. Processing will continue on errors.

Consider below Code:

BulkOperations bulkOps = mongoTemplate.bulkOps(BulkMode.UNORDERED, className);

        for (T entry : entries) {
            DBObject dbDoc = new BasicDBObject();
            mongoTemplate.getConverter().write(entry, dbDoc);
            Update update = Update.fromDBObject(dbDoc);
            bulkOps.updateOne(new Query(Criteria.where("_id").is(dbDoc.get("_id"))), update);
        }
        BulkWriteResult result = bulkOps.execute();

When I execute above code, below exception occurs:

java.lang.IllegalArgumentException: Invalid BSON field name _class
at org.bson.AbstractBsonWriter.writeName(AbstractBsonWriter.java:516)
at com.mongodb.DBObjectCodec.encode(DBObjectCodec.java:129)
at com.mongodb.DBObjectCodec.encode(DBObjectCodec.java:61)
at org.bson.codecs.BsonDocumentWrapperCodec.encode(BsonDocumentWrapperCodec.java:63)
at org.bson.codecs.BsonDocumentWrapperCodec.encode(BsonDocumentWrapperCodec.java:29)
at com.mongodb.connection.UpdateCommandMessage.writeTheWrites(UpdateCommandMessage.java:85)
at com.mongodb.connection.UpdateCommandMessage.writeTheWrites(UpdateCommandMessage.java:43)
at com.mongodb.connection.BaseWriteCommandMessage.encodeMessageBodyWithMetadata(BaseWriteCommandMessage.java:129)
at com.mongodb.connection.RequestMessage.encodeWithMetadata(RequestMessage.java:160)
at com.mongodb.connection.WriteCommandProtocol.sendMessage(WriteCommandProtocol.java:220)
at com.mongodb.connection.WriteCommandProtocol.execute(WriteCommandProtocol.java:101)
at com.mongodb.connection.UpdateCommandProtocol.execute(UpdateCommandProtocol.java:64)
at com.mongodb.connection.UpdateCommandProtocol.execute(UpdateCommandProtocol.java:37)
at com.mongodb.connection.DefaultServer$DefaultServerProtocolExecutor.execute(DefaultServer.java:168)
at com.mongodb.connection.DefaultServerConnection.executeProtocol(DefaultServerConnection.java:289)
at com.mongodb.connection.DefaultServerConnection.updateCommand(DefaultServerConnection.java:143)
at com.mongodb.operation.MixedBulkWriteOperation$Run$3.executeWriteCommandProtocol(MixedBulkWriteOperation.java:490)
at com.mongodb.operation.MixedBulkWriteOperation$Run$RunExecutor.execute(MixedBulkWriteOperation.java:656)
at com.mongodb.operation.MixedBulkWriteOperation$Run.execute(MixedBulkWriteOperation.java:409)
at com.mongodb.operation.MixedBulkWriteOperation$1.call(MixedBulkWriteOperation.java:177)
at com.mongodb.operation.MixedBulkWriteOperation$1.call(MixedBulkWriteOperation.java:168)
at com.mongodb.operation.OperationHelper.withConnectionSource(OperationHelper.java:426)
at com.mongodb.operation.OperationHelper.withConnection(OperationHelper.java:417)
at com.mongodb.operation.MixedBulkWriteOperation.execute(MixedBulkWriteOperation.java:168)
at com.mongodb.operation.MixedBulkWriteOperation.execute(MixedBulkWriteOperation.java:74)
at com.mongodb.Mongo.execute(Mongo.java:845)
at com.mongodb.Mongo$2.execute(Mongo.java:828)
at com.mongodb.DBCollection.executeBulkWriteOperation(DBCollection.java:2309)
at com.mongodb.DBCollection.executeBulkWriteOperation(DBCollection.java:2302)
at com.mongodb.BulkWriteOperation.execute(BulkWriteOperation.java:121)
at org.springframework.data.mongodb.core.DefaultBulkOperations.execute(DefaultBulkOperations.java:278)
at com.nayapay.biller.service.UtilityService.saveInBulk(UtilityService.java:122)
at com.nayapay.biller.service.xxx.xxx(xxx.java:447)
at com.nayapay.biller.service.xxx.xxx(xxx.java:172)
at com.nayapay.biller.controller.xxx.updateBatch(xxx.java:88)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
at java.lang.reflect.Method.invoke(Unknown Source)
at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:205)
at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:133)
at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:97)
at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:827)
at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:738)
at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:85)
at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:967)
at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:901)
at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:970)
at org.springframework.web.servlet.FrameworkServlet.doPost(FrameworkServlet.java:872)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:661)
at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:846)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:742)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:52)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:317)
at org.springframework.security.web.access.intercept.FilterSecurityInterceptor.invoke(FilterSecurityInterceptor.java:127)
at org.springframework.security.web.access.intercept.FilterSecurityInterceptor.doFilter(FilterSecurityInterceptor.java:91)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.access.ExceptionTranslationFilter.doFilter(ExceptionTranslationFilter.java:114)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.session.SessionManagementFilter.doFilter(SessionManagementFilter.java:137)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.authentication.AnonymousAuthenticationFilter.doFilter(AnonymousAuthenticationFilter.java:111)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter.doFilter(SecurityContextHolderAwareRequestFilter.java:170)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.savedrequest.RequestCacheAwareFilter.doFilter(RequestCacheAwareFilter.java:63)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.authentication.logout.LogoutFilter.doFilter(LogoutFilter.java:116)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.header.HeaderWriterFilter.doFilterInternal(HeaderWriterFilter.java:64)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.context.SecurityContextPersistenceFilter.doFilter(SecurityContextPersistenceFilter.java:105)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter.doFilterInternal(WebAsyncManagerIntegrationFilter.java:56)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:331)
at org.springframework.security.web.FilterChainProxy.doFilterInternal(FilterChainProxy.java:214)
at org.springframework.security.web.FilterChainProxy.doFilter(FilterChainProxy.java:177)
at org.springframework.web.filter.DelegatingFilterProxy.invokeDelegate(DelegatingFilterProxy.java:346)
at org.springframework.web.filter.DelegatingFilterProxy.doFilter(DelegatingFilterProxy.java:262)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:197)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:198)
at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:96)
at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:478)
at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:140)
at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:80)
at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:87)
at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:342)
at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:799)
at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:66)
at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:868)
at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1457)
at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)


at java.lang.Thread.run(Unknown Source)
[ERROR] 2017-11-16 10:52:26.904 [http-nio-127.0.0.1-8084-exec-10] ExceptionController - Error occured: Invalid BSON field name _class

If I exclude _id and or _class, same exception will occurs for other fields!!

So my Questions are:

  1. Is MongoRepository's method save(Iteratable list) execute as bulk??
  2. How to Upsert a list in Bulk(without looking if the document already exists)
  3. How to ignore data integrity exceptions like DuplicateKeyException

Note: Document's id is by set by application and not by Database(will help in filtering purpose)

Answer

user9863401 picture user9863401 · Jun 5, 2018

You can try this, hopefully it will work!

final BulkWriteOperation bulkOps = mongoTemplate.getCollection(className).initializeUnorderedBulkOperation();

        entries.stream().filter(entry -> entry != null).forEach(entry -> {
            DBObject dbDoc = new BasicDBObject();
            mongoTemplate.getConverter().write(entry, dbDoc);
            dbDoc.removeField("_id");
            BulkWriteRequestBuilder bulkWriteRequestBuilder = bulkOps.find(criteria.apply(dbDoc).getCriteriaObject());
            BulkUpdateRequestBuilder upsertReq = bulkWriteRequestBuilder.upsert();
            upsertReq.replaceOne(dbDoc);
        });

        BulkWriteResult result = bulkOps.execute();

Related questions

Spring Custom Query with pageable

I want to to implement pagination in spring application.I know using repository we can implement pagination but we can not write our own query for data retrieve there are limited methods in repository that too there is no method …

Read More
How to configure a MongoDB cluster which supports sessions?

I want to explore the new transaction feature of MongoDB and use Spring Data MongoDB. However, I get the exception message "Sessions are not supported by the MongoDB cluster to which this client is connected". Any hint regarding the config …

Read More
Spring Boot and how to configure connection details to MongoDB?

Being new to Spring Boot I am wondering on how I can configure connection details for MongoDB. I have tried the normal examples but none covers the connection details. I want to specify the database that is going to be …

Read More