bugfix: bid >= 0

Author: davies

core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala

@@ -211,7 +211,7 @@ private[spark] class PythonRDD(
         for (bid <- bids) {
           if (!nbids.contains(bid)) {
             // remove the broadcast from worker
-            dataOut.writeLong(-bid)
+            dataOut.writeLong(- bid - 1)  // bid >= 0
             bids.remove(bid)
           }
         }

python/pyspark/worker.py

@@ -69,11 +69,12 @@ def main(infile, outfile):
         ser = CompressedSerializer(pickleSer)
         for _ in range(num_broadcast_variables):
             bid = read_long(infile)
-            if bid > 0:
+            if bid >= 0:
                 value = ser._read_with_length(infile)
                 _broadcastRegistry[bid] = Broadcast(bid, value)
             else:
-                _broadcastRegistry.pop(-bid, None)
+                bid = - bid - 1
+                _broadcastRegistry.pop(bid, None)
 
         command = pickleSer._read_with_length(infile)
         (func, deserializer, serializer) = command