[BUG] Tuplex crashes in exception resolver when take/collect() is called twice

The tuplex crashes from a corrupted exception partition when the pipeline is executed twice. This could be an issue from invalidating partition.

Example pipeline when it crashes:
```
ds = c.parallelize([(1, "A"),(2, "a"),(3, 2)]).filter(lambda a, b: a > 1)
ds.collect()
ds.collect()
```