def check_input_persisted_in_s3(**kwargs):
    """
        Fetch differnt versions inside prefix
        And check if path created for input hour.
        If present, mark step as success. if not, raise exception for input data unavailability.
    """
    exec_datehour = datetime.strptime(kwargs['ts_nodash'], '%Y%m%dT%H%M%S')

    validation_configs = kwargs['input-validation-step-config']
    step_conf = Variable.get(validation_configs, deserialize_json=True)
    input_bucket = step_conf['input_bucket']
    key_prefix = step_conf['key_prefix']
    input_hours_to_skip_for_alert = step_conf['expected_hours_to_skip_for_alert'].split(",")

    aws_region = Variable.get('aws_region').lower()

    s3_client = boto3.client('s3', aws_region)

    versions_list = list_folders(s3_client, input_bucket, key_prefix)
    input_path_versions_datehour = list(
        map(lambda str: str + f"{exec_datehour.date()}/"f"{exec_datehour.hour:02d}", versions_list))

    paths_present = []
    for prefix in input_path_versions_datehour:
        try:
            resp = s3_client.list_objects(Bucket=input_bucket, Prefix=prefix, Delimiter='/', MaxKeys=1)
            if 'CommonPrefixes' in resp:
                paths_present.append(prefix)
        except ClientError as exc:
            LOG.info(f"Path not found for {prefix}: ", exc)

    if '{:02d}'.format(exec_datehour.hour) not in input_hours_to_skip_for_alert:
        if len(paths_present) > 0:
            LOG.info(f"Input path exists for the hour {exec_datehour} : {paths_present}")
        else:
            raise Exception(f"Input path does not exist for the hour: {exec_datehour} in s3://{input_bucket}/{key_prefix}")
    else:
        LOG.info(f"Skipped check for the hour {exec_datehour}")