class StagingItemReader<T>
		implements ItemReader<ProcessIndicatorItemWrapper<T>>, StepExecutionListener, InitializingBean, DisposableBean {

	private static final Log logger = LogFactory.getLog(StagingItemReader.class);

	private StepExecution stepExecution;

	private final Lock lock = new ReentrantLock();

	private volatile boolean initialized = false;

	private volatile Iterator<Long> keys;

	private JdbcOperations jdbcTemplate;

	public void setDataSource(DataSource dataSource) {
		jdbcTemplate = new JdbcTemplate(dataSource);
	}

	@Override
	public void destroy() throws Exception {
		initialized = false;
		keys = null;
	}

	@Override
	public final void afterPropertiesSet() throws Exception {
		Assert.state(jdbcTemplate != null, "You must provide a DataSource.");
	}

	private List<Long> retrieveKeys() {<FILL_FUNCTION_BODY>}

	@Nullable
	@Override
	public ProcessIndicatorItemWrapper<T> read() {
		if (!initialized) {
			throw new ReaderNotOpenException("Reader must be open before it can be used.");
		}

		Long id = null;
		synchronized (lock) {
			if (keys.hasNext()) {
				id = keys.next();
			}
		}
		if (logger.isDebugEnabled()) {
			logger.debug("Retrieved key from list: " + id);
		}
		if (id == null) {
			return null;
		}
		@SuppressWarnings("unchecked")
		T result = (T) jdbcTemplate.queryForObject("SELECT VALUE FROM BATCH_STAGING WHERE ID=?",
				(rs, rowNum) -> deserialize(rs.getBinaryStream(1)), id);

		return new ProcessIndicatorItemWrapper<>(id, result);
	}

	private static Object deserialize(InputStream inputStream) {
		if (inputStream == null) {
			return null;
		}
		try (var objectInputStream = new ObjectInputStream(inputStream)) {
			return objectInputStream.readObject();
		}
		catch (Exception e) {
			throw new IllegalArgumentException("Failed to deserialize object", e);
		}
	}

	@Nullable
	@Override
	public ExitStatus afterStep(StepExecution stepExecution) {
		return null;
	}

	@Override
	public void beforeStep(StepExecution stepExecution) {
		this.stepExecution = stepExecution;
		synchronized (lock) {
			if (keys == null) {
				keys = retrieveKeys().iterator();
				logger.info("Keys obtained for staging.");
				initialized = true;
			}
		}
	}

}


		this.lock.lock();
		try {

			return jdbcTemplate.query(

					"SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID",

					(rs, rowNum) -> rs.getLong(1),

					stepExecution.getJobExecution().getJobId(), StagingItemWriter.NEW);

		}
		finally {
			this.lock.unlock();
		}

	
<no_super_class>