Clustering : ConflictingModificationAppliedException on heavy clustering datastore use

When some component requests "too many operations in too short time", the clustering datastore fails with this exception:

2015-05-25 15:01:40,611 | WARN | ult-dispatcher-3 | ConcurrentDOMDataBroker | 212 - org.opendaylight.controller.sal-distributed-datastore - 1.2.0.SNAPSHOT | Tx: DOM-4 Error during phase CAN_COMMIT, starting Abort
org.opendaylight.yangtools.yang.data.api.schema.tree.ConflictingModificationAppliedException: Node was replaced by other transaction.
at org.opendaylight.yangtools.yang.data.impl.schema.tree.SchemaAwareApplyOperation.checkConflicting(SchemaAwareApplyOperation.java:69)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.yangtools.yang.data.impl.schema.tree.SchemaAwareApplyOperation.checkNotConflicting(SchemaAwareApplyOperation.java:98)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.yangtools.yang.data.impl.schema.tree.SchemaAwareApplyOperation.checkWriteApplicable(SchemaAwareApplyOperation.java:168)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.yangtools.yang.data.impl.schema.tree.SchemaAwareApplyOperation.checkApplicable(SchemaAwareApplyOperation.java:127)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.yangtools.yang.data.impl.schema.tree.AbstractNodeContainerModificationStrategy.checkChildPreconditions(AbstractNodeContainerModificationStrategy.java:205)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.yangtools.yang.data.impl.schema.tree.AbstractNodeContainerModificationStrategy.checkTouchApplicable(AbstractNodeContainerModificationStrategy.java:189)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.yangtools.yang.data.impl.schema.tree.SchemaAwareApplyOperation.checkApplicable(SchemaAwareApplyOperation.java:124)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.yangtools.yang.data.impl.schema.tree.RootModificationApplyOperation.checkApplicable(RootModificationApplyOperation.java:71)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.yangtools.yang.data.impl.schema.tree.AbstractDataTreeTip.validate(AbstractDataTreeTip.java:37)[118:org.opendaylight.yangtools.yang-data-impl:0.7.0.SNAPSHOT]
at org.opendaylight.controller.cluster.datastore.SimpleShardDataTreeCohort.canCommit(SimpleShardDataTreeCohort.java:39)[212:org.opendaylight.controller.sal-distributed-datastore:1.2.0.SNAPSHOT]
at org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.doCanCommit(ShardCommitCoordinator.java:301)[212:org.opendaylight.controller.sal-distributed-datastore:1.2.0.SNAPSHOT]
at org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.handleCanCommit(ShardCommitCoordinator.java:259)[212:org.opendaylight.controller.sal-distributed-datastore:1.2.0.SNAPSHOT]
at org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.handleReadyLocalTransaction(ShardCommitCoordinator.java:231)[212:org.opendaylight.controller.sal-distributed-datastore:1.2.0.SNAPSHOT]
at org.opendaylight.controller.cluster.datastore.Shard.handleReadyLocalTransaction(Shard.java:453)[212:org.opendaylight.controller.sal-distributed-datastore:1.2.0.SNAPSHOT]
at org.opendaylight.controller.cluster.datastore.Shard.onReceiveCommand(Shard.java:237)[212:org.opendaylight.controller.sal-distributed-datastore:1.2.0.SNAPSHOT]
at akka.persistence.UntypedPersistentActor.onReceive(Eventsourced.scala:430)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at org.opendaylight.controller.cluster.common.actor.MeteringBehavior.apply(MeteringBehavior.java:97)[204:org.opendaylight.controller.sal-clustering-commons:1.2.0.SNAPSHOT]
at akka.actor.ActorCell$$anonfun$become$1.applyOrElse(ActorCell.scala:534)[197:com.typesafe.akka.actor:2.3.10]
at akka.persistence.Recovery$State$class.process(Recovery.scala:30)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.persistence.ProcessorImpl$$anon$2.process(Processor.scala:103)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.persistence.ProcessorImpl$$anon$2.aroundReceive(Processor.scala:114)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.persistence.Recovery$class.aroundReceive(Recovery.scala:265)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.persistence.UntypedPersistentActor.akka$persistence$Eventsourced$$super$aroundReceive(Eventsourced.scala:428)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.persistence.Eventsourced$$anon$2.doAroundReceive(Eventsourced.scala:82)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.persistence.Eventsourced$$anon$2.aroundReceive(Eventsourced.scala:78)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.persistence.Eventsourced$class.aroundReceive(Eventsourced.scala:369)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.persistence.UntypedPersistentActor.aroundReceive(Eventsourced.scala:428)[202:com.typesafe.akka.persistence.experimental:2.3.10]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)[197:com.typesafe.akka.actor:2.3.10]
at akka.actor.ActorCell.invoke(ActorCell.scala:487)[197:com.typesafe.akka.actor:2.3.10]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)[197:com.typesafe.akka.actor:2.3.10]
at akka.dispatch.Mailbox.run(Mailbox.scala:221)[197:com.typesafe.akka.actor:2.3.10]
at akka.dispatch.Mailbox.exec(Mailbox.scala:231)[197:com.typesafe.akka.actor:2.3.10]
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)[194:org.scala-lang.scala-library:2.10.4.v20140209-180020-VFINAL-b66a39653b]
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)[194:org.scala-lang.scala-library:2.10.4.v20140209-180020-VFINAL-b66a39653b]
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)[194:org.scala-lang.scala-library:2.10.4.v20140209-180020-VFINAL-b66a39653b]
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)[194:org.scala-lang.scala-library:2.10.4.v20140209-180020-VFINAL-b66a39653b]
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)[194:org.scala-lang.scala-library:2.10.4.v20140209-180020-VFINAL-b66a39653b]

An easy way to reproduce the problem is to switch on topology in the BGP module and then push some 4000 prefixes into it (use this tool to do that: https://git.opendaylight.org/gerrit/#/c/19603/4/test/tools/fastbgp/play.py). Also increase akka.timeout to something like 5 minutes so you don't get bothered with akka timeouts.