// MongoStepExecutionDao.java (Line 165-177)
@Override
public long countStepExecutions(JobInstance jobInstance, String stepName) {
    Query query = query(where("jobInstanceId").is(jobInstance.getId()));
    List jobExecutions = this.mongoOperations.find(...);
    return this.mongoOperations.count(
            query(where("jobExecutionId").in(jobExecutions.stream()
                .map(JobExecution::getJobExecutionId)
                .toList())),  // stepName parameter is ignored
            StepExecution.class, STEP_EXECUTIONS_COLLECTION_NAME);
}

// SimpleStepHandler.java (Line 205-221)
long stepExecutionCount = jobRepository.getStepExecutionCount(jobInstance, step.getName());
if (stepExecutionCount < step.getStartLimit()) {
    return true;
} else {
    throw new StartLimitExceededException(...);
}

@Test
void testCountStepExecutionsFiltersByStepName() {
    // given
    dao.createStepExecution("stepA", jobExecution);
    dao.createStepExecution("stepA", jobExecution);
    dao.createStepExecution("stepB", jobExecution);
    dao.createStepExecution("stepC", jobExecution);

    // when
    long countA = dao.countStepExecutions(jobInstance, "stepA");
    long countB = dao.countStepExecutions(jobInstance, "stepB");
    long countC = dao.countStepExecutions(jobInstance, "stepC");
    long countNonExistent = dao.countStepExecutions(jobInstance, "nonExistentStep");

    // then
    assertEquals(2, countA);  // Bug: returns 4
    assertEquals(1, countB);  // Bug: returns 4
    assertEquals(1, countC);  // Bug: returns 4
    assertEquals(0, countNonExistent);  // Bug: returns 4
}

return this.mongoOperations.count(
        query(where("jobExecutionId").in(jobExecutions.stream()
            .map(JobExecution::getJobExecutionId)
            .toList())
            .and("name").is(stepName)),  // Added stepName filter
        StepExecution.class, STEP_EXECUTIONS_COLLECTION_NAME);