code: def _attempt_shard_retry(self, shard_state, tstate):
      
      shard_attempts = shard_state.retries + 1
  
      if shard_attempts >= parameters.config.SHARD_MAX_ATTEMPTS:
        logging.warning(
            "Shard attempt %s exceeded %s max attempts.",
            shard_attempts, parameters.config.SHARD_MAX_ATTEMPTS)
        return self._TASK_DIRECTIVE.FAIL_TASK
      if tstate.output_writer and (
          not tstate.output_writer._supports_shard_retry(tstate)):
        logging.warning("Output writer %s does not support shard retry.",
                        tstate.output_writer.__class__.__name__)
        return self._TASK_DIRECTIVE.FAIL_TASK
  
      shard_state.reset_for_retry()
      logging.warning("Shard %s attempt %s failed with up to %s attempts.",
                      shard_state.shard_id,
                      shard_state.retries,
                      parameters.config.SHARD_MAX_ATTEMPTS)
      output_writer = None
      if tstate.output_writer:
        output_writer = tstate.output_writer.create(
            tstate.mapreduce_spec, shard_state.shard_number, shard_attempts + 1)
      tstate.reset_for_retry(output_writer)
      return self._TASK_DIRECTIVE.RETRY_SHARD
docs: Whether to retry shard.
  
      This method may modify shard_state and tstate to prepare for retry or fail.
  
      Args:
        shard_state: model.ShardState for current shard.
        tstate: model.TransientShardState for current shard.
  
      Returns:
        A _TASK_DIRECTIVE enum. RETRY_SHARD if shard should be retried.
      FAIL_TASK otherwise.