class DbStateBackend extends AbstractStateBackend {

	@Override
	public void initializeForJob(final Environment env,
		String operatorIdentifier,
		TypeSerializer<?> keySerializer) throws Exception {
		super.initializeForJob(env, operatorIdentifier, keySerializer);

		this.operatorIdentifier = operatorIdentifier;

		this.rnd = new Random();
		this.env = env;
		this.jobId = env.getJobID().toString().substring(0, 16);

		connections = dbConfig.createShardedConnection();

		// We want the most light-weight transaction isolation level as we don't
		// have conflicting reads/writes. We just want to be able to roll back
		// batch inserts for k-v snapshots. This requirement might be removed in
		// the future.
		connections.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);

		// If we have a different backend for non-partitioned states we
		// initialize that, otherwise create tables for storing the checkpoints.
		//
		// Currently all non-partitioned states are written to the first
		// database shard
		if (nonPartitionedStateBackend == null) {
			insertStatement = retry(new Callable<PreparedStatement>() {
				public PreparedStatement call() throws SQLException {
					dbAdapter.createCheckpointsTable(jobId, getConnections().getFirst());
					return dbAdapter.prepareCheckpointInsert(jobId,
							getConnections().getFirst());
				}
			}, numSqlRetries, sqlRetrySleep);
		} else {
			nonPartitionedStateBackend.initializeForJob(env, operatorIdentifier, keySerializer);
		}

		if (LOG.isDebugEnabled()) {
			LOG.debug("Database state backend successfully initialized");
		}
	}

	public  DbStateBackend(DbBackendConfig backendConfig);
	public  DbStateBackend(DbBackendConfig backendConfig, AbstractStateBackend backend);

}