[Spring Batch] 메타데이터 테이블과 시퀀스

 

 

 

org.springframework.dao.DataAccessResourceFailureException: Could not obtain sequence value

 

 

스프링배치로 애플리케이션을 만들고 실행할 때 실행 관련 상태를 저장하는 메타데이터 테이블을 관리하게 된다.

 

시스템에서 SQL Server를 사용하는 경우 메타데이터 테이블에 값을 넣을 때 시퀀스 객체를 사용하는데.. 해당 DBMS에서 시퀀스 객체를 지원하지 않는 경우 위와 같은 오류가 발생한다.

 

 

소스코드를 까보면 DefaultDataFieldMaxValueIncrementerFactory 클래스를 확인할 수 있다.

 

public class DefaultDataFieldMaxValueIncrementerFactory implements DataFieldMaxValueIncrementerFactory {

	private final DataSource dataSource;

	private String incrementerColumnName = "ID";

	/**
	 * Public setter for the column name (defaults to "ID") in the incrementer. Only used
	 * by some platforms (Derby, HSQL, MySQL, SQL Server and Sybase), and should be fine
	 * for use with Spring Batch meta data as long as the default batch schema hasn't been
	 * changed.
	 * @param incrementerColumnName the primary key column name to set
	 */
	public void setIncrementerColumnName(String incrementerColumnName) {
		this.incrementerColumnName = incrementerColumnName;
	}

	public DefaultDataFieldMaxValueIncrementerFactory(DataSource dataSource) {
		this.dataSource = dataSource;
	}

	@Override
	public DataFieldMaxValueIncrementer getIncrementer(String incrementerType, String incrementerName) {
		DatabaseType databaseType = DatabaseType.valueOf(incrementerType.toUpperCase());

		if (databaseType == DB2 || databaseType == DB2AS400) {
			return new Db2LuwMaxValueIncrementer(dataSource, incrementerName);
		}
		else if (databaseType == DB2ZOS) {
			return new Db2MainframeMaxValueIncrementer(dataSource, incrementerName);
		}
		else if (databaseType == DERBY) {
			return new DerbyMaxValueIncrementer(dataSource, incrementerName, incrementerColumnName);
		}
		else if (databaseType == HSQL) {
			return new HsqlMaxValueIncrementer(dataSource, incrementerName, incrementerColumnName);
		}
		else if (databaseType == H2) {
			return new H2SequenceMaxValueIncrementer(dataSource, incrementerName);
		}
		else if (databaseType == HANA) {
			return new HanaSequenceMaxValueIncrementer(dataSource, incrementerName);
		}
		else if (databaseType == MYSQL) {
			MySQLMaxValueIncrementer mySQLMaxValueIncrementer = new MySQLMaxValueIncrementer(dataSource,
					incrementerName, incrementerColumnName);
			mySQLMaxValueIncrementer.setUseNewConnection(true);
			return mySQLMaxValueIncrementer;
		}
		else if (databaseType == MARIADB) {
			return new MariaDBSequenceMaxValueIncrementer(dataSource, incrementerName);
		}
		else if (databaseType == ORACLE) {
			return new OracleSequenceMaxValueIncrementer(dataSource, incrementerName);
		}
		else if (databaseType == POSTGRES) {
			return new PostgresSequenceMaxValueIncrementer(dataSource, incrementerName);
		}
		else if (databaseType == SQLITE) {
			return new SqliteMaxValueIncrementer(dataSource, incrementerName, incrementerColumnName);
		}
		else if (databaseType == SQLSERVER) {
			return new SqlServerSequenceMaxValueIncrementer(dataSource, incrementerName);
		}
		else if (databaseType == SYBASE) {
			return new SybaseMaxValueIncrementer(dataSource, incrementerName, incrementerColumnName);
		}
		throw new IllegalArgumentException("databaseType argument was not on the approved list");
	}

	@Override
	public boolean isSupportedIncrementerType(String incrementerType) {
		for (DatabaseType type : DatabaseType.values()) {
			if (type.name().equalsIgnoreCase(incrementerType)) {
				return true;
			}
		}

		return false;
	}

	@Override
	public String[] getSupportedIncrementerTypes() {

		List<String> types = new ArrayList<>();

		for (DatabaseType type : DatabaseType.values()) {
			types.add(type.name());
		}

		return types.toArray(new String[types.size()]);
	}

}

 

 

getIncrementer 메서드를 확인해보면 시스템에서 사용하는 DBMS에 따라 다른 incrementer를 반환하고 있음을 확인할 수 있다.

 

 

public class SqlServerSequenceMaxValueIncrementer extends AbstractSequenceMaxValueIncrementer {

	/**
	 * Default constructor for bean property style usage.
	 * @see #setDataSource
	 * @see #setIncrementerName
	 */
	public SqlServerSequenceMaxValueIncrementer() {
	}

	/**
	 * Convenience constructor.
	 * @param dataSource the DataSource to use
	 * @param incrementerName the name of the sequence to use
	 */
	public SqlServerSequenceMaxValueIncrementer(DataSource dataSource, String incrementerName) {
		super(dataSource, incrementerName);
	}

	@Override
	protected String getSequenceQuery() {
		return "select next value for " + getIncrementerName();
	}

}

 

 

SQL Server를 사용하는 경우 SqlServerSequenceMaxValueIncrementer 클래스를 사용하는데.. 
해당 부분을 살펴보면 "SELECT NEXT VALUE FOR" 구문으로 DBMS의 시퀀스 객체를 사용하는 쿼리를 확인할 수 있다.

 

SQL Server 2012 버전부터 도입됐고, 스프링배치에서는 사용하는 DBMS만 확인할 뿐 DBMS의 버전을 따로 확인하지는 않는다.

SQL Server 2005 버전을 사용하는 경우 시퀀스를 지원하지 않아 메타데이터 테이블에 값을 넣을 때 오류가 발생한다. 

 

Linked Server 를 사용해서 다른 데이터베이스에 메타데이터 테이블을 구축하는 방법도 있지만..

시스템에서 해당 부분만 수정해서 시퀀스를 지원하지 않는 DBMS에서도 스프링 배치를 사용할 수 있도록 설정하자. 

 

 

public class CustomDefaultDataFieldMaxValueIncrementerFactory extends DefaultDataFieldMaxValueIncrementerFactory{
    private final DataSource dataSource;

    public CustomDefaultDataFieldMaxValueIncrementerFactory(DataSource dataSource) {
        super(dataSource);
        this.dataSource = dataSource;
    }
    @Override
    public DataFieldMaxValueIncrementer getIncrementer(String databaseType, String incrementerName) {
        CustomIncrementer customIncrementer = new CustomIncrementer(this.dataSource, incrementerName);
        return customIncrementer;
    }

}

 

 

우선 해당 클래스를 상속해 새로 구현한 Incrementer를 사용하도록 설정하자. 

 

public class CustomIncrementer extends AbstractColumnMaxValueIncrementer {

    public CustomIncrementer(DataSource dataSource, String incrementerName) {
        super(dataSource, incrementerName, "ID");
    }

    @Override
    protected long getNextKey() {
        long nextKey = 0;
        try (Connection conn = this.getDataSource().getConnection()) {
            PreparedStatement psSelect = conn.prepareStatement("SELECT ID FROM " + getIncrementerName() + " WITH (UPDLOCK, HOLDLOCK)");
            ResultSet rs = psSelect.executeQuery();
            if (rs.next()) {
                nextKey = rs.getLong(1);
            }
            rs.close();
            psSelect.close();

            PreparedStatement psUpdate = conn.prepareStatement("UPDATE " + getIncrementerName() + " SET ID = ?");
            psUpdate.setLong(1, nextKey + 1);
            psUpdate.executeUpdate();
            psUpdate.close();
        } catch (SQLException e) {
            throw new RuntimeException("Could not get next key for " + getIncrementerName(), e);
        }
        return nextKey + 1;
    }

}

 

 

시퀀스 대신 테이블을 시퀀스 처럼 사용하자.

 

 

 

 

DBMS에서 사용하는 시퀀스와 똑같은 이름을 가진 테이블을 만들고, 그 테이블에 column과 row를 하나만 만들어준다.

그 값 하나를 1씩 증가시키면서 사용해 시퀀스처럼 사용하자. (BATCH_JOB_SEQ)

 

시퀀스 객체를 직접 사용하는 것 보다 동시성, 무결성은 떨어질 수 있지만..

이렇게 설정해주면 시퀀스 객체를 지원하지 않는 버전에서도 스프링 배치의 메타데이터 테이블을 사용할 수 있다.