title: Spark accumulators thread-safety (LongAccumulator seems to be broken)
body: <p>I'm writing a custom <code>AccumulatorV2</code> and want it to be properly synchronized.</p>
  
  <p>I've read a few post about the Spark <code>AccumulatorV2</code> concurrency problems and as far as I understand</p>
  
  <p>1) Writes to an accumulator happens before reads and vise versa</p>
  
  <p>2) But writes to an accumulator from different partitions are not synchronized</p>
  
  <p>So accumulator's <code>add</code> method must support concurrent access.</p>
  
  <p>Here is the proof:</p>
  
  <pre><code>object MyAccumulator {
    val uniqueId: AtomicInteger = new AtomicInteger()
  }
  
  //Just a regular int accumulator but with a delay inside Add method
  class MyAccumulator(var w: Int = 0) extends AccumulatorV2[Int, Int] {
  
    //log this id to prove that updates happen on the same object
    val thisAccId = MyAccumulator.uniqueId.incrementAndGet()
  
    override def isZero: Boolean = w == 0
  
    override def copy(): AccumulatorV2[Int, Int] = new MyAccumulator(w)
  
    override def reset(): Unit = w = 0
  
    override def add(v: Int): Unit = {
      println(s"Start adding $thisAccId " + Thread.currentThread().getId)
      Thread.sleep(500)
      w += v
      println(s"End adding $thisAccId " + Thread.currentThread().getId)
    }
  
    override def merge(other: AccumulatorV2[Int, Int]): Unit = w += other.value
  
    override def value: Int = w
  }
  
  object Test extends App {
    val conf = new SparkConf()
    conf.setMaster("local[5]")
    conf.setAppName("test")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(1 to 50, 10)
    val acc = new MyAccumulator
    sc.register(acc)
    rdd.foreach(x =&gt; acc.add(x))
    println(acc.value)
  
    sc.stop()
  }
  </code></pre>
  
  <p>Output example: </p>
  
  <pre><code>Start adding 1 73
  Start adding 1 77
  Start adding 1 76
  Start adding 1 74
  Start adding 1 75
  End adding 1 74
  End adding 1 75
  Start adding 1 74
  End adding 1 76
  End adding 1 77
  End adding 1 73
  Start adding 1 77
  Start adding 1 76
  ....
  ....
  ....
  1212
  </code></pre>
  
  <p>We can see that there are several threads are inside the <code>add</code> method concurrently and the result is wrong (it must be 1275).</p>
  
  <p>Then I looked into the source code of built-in accumulators (e.g. <code>org.apache.spark.util.LongAccumulator</code>) and found no trace of synchronization. It uses just mutable <code>var</code>s.</p>
  
  <p>How can it work at all?</p>
  
  <p>UPDATE:
  LongAccumulator is broken indeed (the following code fails) :</p>
  
  <pre><code>object LongAccumulatorIsBroken extends App {
    val conf = new SparkConf()
    conf.setMaster("local[5]")
    conf.setAppName("test")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize((1 to 50000).map(x =&gt; 1), 1000)
    val acc = new LongAccumulator
    sc.register(acc)
    rdd.foreach(x =&gt; acc.add(x))
    val value = acc.value
    println(value)
    sc.stop()
    assert(value == 50000, message = s"$value is not 50000")
  }
  </code></pre>