docstr: Verify the attempt numbers are correctly reported.
docstr_tok: ["Verify","the","attempt","numbers","are","correctly","reported","."]
code: def test_attempt_number(self):
      """Verify the attempt numbers are correctly reported."""
      rdd = self.sc.parallelize(range(10))
      # Verify a simple job with no failures
      attempt_numbers = rdd.map(
          lambda x: TaskContext.get().attemptNumber()
      ).collect()
      map(lambda attempt: self.assertEqual(0, attempt), attempt_numbers)
  
      def fail_on_first(x):
          """Fail on the first attempt so we get a positive attempt number"""
          tc = TaskContext.get()
          attempt_number = tc.attemptNumber()
          partition_id = tc.partitionId()
          attempt_id = tc.taskAttemptId()
          if attempt_number == 0 and partition_id == 0:
              raise RuntimeError("Failing on first attempt")
          else:
              return [x, partition_id, attempt_number, attempt_id]
  
      result = rdd.map(fail_on_first).collect()
      # We should re-submit the first partition to it but other partitions should be attempt 0
      self.assertEqual([0, 0, 1], result[0][0:3])
      self.assertEqual([9, 3, 0], result[9][0:3])
      first_partition = filter(lambda x: x[1] == 0, result)
      map(lambda x: self.assertEqual(1, x[2]), first_partition)
      other_partitions = filter(lambda x: x[1] != 0, result)
      map(lambda x: self.assertEqual(0, x[2]), other_partitions)
      # The task attempt id should be different
      self.assertTrue(result[0][3] != result[9][3])
  
imports: import os
  import random
  import stat
  import sys
  import tempfile
  import time
  import unittest
  from pyspark import SparkConf, SparkContext, TaskContext, BarrierTaskContext
  from pyspark.testing.utils import PySparkTestCase, SPARK_HOME, eventually
label_desc: Use image node_id 5 for calling the TaskContextTests obj's underlying member method code with example usage: obj.test_attempt_number() and returns: list