StopIteration in pyspark mapper gives partial results

Consider the following code

def mapper(xx):
     if xx % 2 == 0:
         raise StopIteration()
     else:
         return xx
sc.parallelize(range(100)).map(mapper).collect()

The result I get is [57, 71, 85]

I think it happens because map is implemented in terms of mapPartitionsWithIndex using a custom iterator, so the StopIteration raised by the mapper is handled by that iterator. I think this should be raised to the user instead.

I think I can take care of this, if I am allowed to (first time I contribute, not sure how it works)

NB: this may be the underlying cause of https://issues.apache.org/jira/browse/SPARK-23754