Spark accumulators thread-safety (LongAccumulator seems to be broken)

I'm writing a custom AccumulatorV2 and want it to be properly synchronized.

I've read a few post about the Spark AccumulatorV2 concurrency problems and as far as I understand

1) Writes to an accumulator happens before reads and vise versa

2) But writes to an accumulator from different partitions are not synchronized

So accumulator's add method must support concurrent access.

Here is the proof:

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 => acc.add(x))
  println(acc.value)

  sc.stop()
}

Output example:

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

We can see that there are several threads are inside the add method concurrently and the result is wrong (it must be 1275).

Then I looked into the source code of built-in accumulators (e.g. org.apache.spark.util.LongAccumulator) and found no trace of synchronization. It uses just mutable vars.

How can it work at all?

UPDATE: LongAccumulator is broken indeed (the following code fails) :

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 => 1), 1000)
  val acc = new LongAccumulator
  sc.register(acc)
  rdd.foreach(x => acc.add(x))
  val value = acc.value
  println(value)
  sc.stop()
  assert(value == 50000, message = s"$value is not 50000")
}