InMemoryDatabase 에 세이브포인트를 추가하세요. savepoint() 는 현재 pending 의 복사본을 스택에 저장하고, rollbackToSavepoint() 는 마지막 세이브포인트 상태로 pending 을 되돌립니다(트랜잭션은 유지). 이를 이용해 "청크 안에서 건별로 세이브포인트를 잡고, 실패 건만 되돌린 뒤 나머지를 커밋"하는 코드를 작성하세요.
// InMemoryDatabase 에 추가
private final Deque<Map<String, Long>> savepoints = new ArrayDeque<>();
public void savepoint() {
requireTx();
savepoints.push(new HashMap<>(pending));
}
public void rollbackToSavepoint() {
requireTx();
if (savepoints.isEmpty()) throw new IllegalStateException("세이브포인트 없음");
pending = savepoints.pop();
}
// commit()/rollback() 에서 savepoints.clear() 추가
// 사용: 실패 건만 제외하고 커밋
db.begin();
int ok = 0, failed = 0;
for (Transfer t : chunk) {
db.savepoint();
try {
apply(t);
ok++;
} catch (InsufficientBalanceException e) {
db.rollbackToSavepoint(); // 이 건의 변경만 취소
failed++;
System.out.println(" skip " + t.txId() + ": " + e.getMessage());
}
}
db.commit();
System.out.printf("커밋: 성공 %d건, 제외 %d건%n", ok, failed);
// 출력:
// skip TX-000350: ACC-04 잔액 1009000 < 이체 999999999
// 커밋: 성공 99건, 제외 1건
// JDBC 대응: conn.setSavepoint() / conn.rollback(savepoint)TransferBatch 의 체크포인트를 파일이 아니라 InMemoryDatabase 안에 저장하도록 바꾸세요(키 checkpoint:{jobName}:line, checkpoint:{jobName}:chunk). 체크포인트 저장을 db.commit() 앞에 두어 데이터와 같은 트랜잭션으로 커밋되게 하고, 이렇게 하면 2.4 절의 "불일치 창"이 왜 사라지는지 주석으로 설명하세요.
// run() 시작부: 체크포인트 로드
long skipLines = db.getOrDefault("checkpoint:" + jobName + ":line", 0);
int chunkNo = (int) db.getOrDefault("checkpoint:" + jobName + ":chunk", 0);
// 청크 처리부
db.begin();
try {
for (Transfer t : chunk) { ... apply(t); ... }
db.put("checkpoint:" + jobName + ":line", lineNo); // 같은 트랜잭션 안
db.put("checkpoint:" + jobName + ":chunk", chunkNo);
db.commit(); // 데이터 + 체크포인트가 한 번에 확정
// 커밋 전에 죽으면: 데이터도 체크포인트도 없음 → 재시작 시 이 청크 처음부터 (정상)
// 커밋 후에 죽으면: 데이터도 체크포인트도 있음 → 재시작 시 다음 청크부터 (정상)
// "데이터는 있는데 체크포인트는 없는" 상태가 존재할 수 없다 = 불일치 창 소멸
} catch (InsufficientBalanceException e) {
db.rollback();
db.begin(); // 롤백 청크도 전진시키려면 별도 tx 로 체크포인트만
db.put("checkpoint:" + jobName + ":line", lineNo);
db.put("checkpoint:" + jobName + ":chunk", chunkNo);
db.commit();
}
// 정상 종료 시
db.begin();
db.put("checkpoint:" + jobName + ":line", 0);
db.put("checkpoint:" + jobName + ":chunk", 0);
db.commit();
// 출력: 파일 방식과 동일한 로그. 단 done: 마킹 없이도 이중 처리가 발생하지 않음
// (그래도 done: 마킹은 유지할 것 — 입력 파일이 바뀌거나 수동 재실행 시 안전망)파일 출력에 청크 커밋을 구현하세요. ChunkFileWriter 는 청크마다 flush() 하고 "지금까지 쓴 바이트 수"를 체크포인트에 기록합니다. 재시작 시 FileChannel.truncate(bytes) 로 마지막 커밋 위치까지 잘라낸 뒤 append 모드로 이어 씁니다. 쓰기 도중 예외로 반쯤 쓰인 청크가 있는 상황을 만들고, 재시작 후 파일 줄 수가 정확한지 검증하세요.
class ChunkFileWriter implements AutoCloseable {
private final Path file, cpFile;
private final BufferedWriter w;
private long committedBytes;
ChunkFileWriter(Path file, Path cpFile) throws IOException {
this.file = file; this.cpFile = cpFile;
committedBytes = Files.exists(cpFile) ? Long.parseLong(Files.readString(cpFile).trim()) : 0;
if (Files.exists(file)) {
try (FileChannel ch = FileChannel.open(file, StandardOpenOption.WRITE)) {
ch.truncate(committedBytes); // 미커밋 꼬리 제거
}
}
w = Files.newBufferedWriter(file, StandardCharsets.UTF_8,
StandardOpenOption.CREATE, StandardOpenOption.APPEND);
}
void writeChunk(List<String> lines) throws IOException {
for (String l : lines) { w.write(l); w.newLine(); }
w.flush(); // OS 로 내보냄
committedBytes = Files.size(file);
Files.writeString(cpFile, Long.toString(committedBytes)); // 커밋 포인트 기록
}
long committedBytes() { return committedBytes; }
@Override public void close() throws IOException { w.close(); }
}
// 테스트: 3청크 쓰고, 4번째 청크 중간에 죽은 것처럼 flush 없이 일부만 쓰기
Path out = Path.of("data/chunked.txt"), cp = Path.of("data/chunked.cp");
Files.deleteIfExists(out); Files.deleteIfExists(cp);
try (ChunkFileWriter cw = new ChunkFileWriter(out, cp)) {
for (int c = 0; c < 3; c++) cw.writeChunk(List.of("chunk" + c + "-a", "chunk" + c + "-b"));
}
Files.writeString(out, "partial-garbage\n", StandardCharsets.UTF_8, StandardOpenOption.APPEND); // 반쯤 쓰인 청크
System.out.println("장애 후 줄 수: " + Files.readAllLines(out).size()); // 7
try (ChunkFileWriter cw = new ChunkFileWriter(out, cp)) { // 재시작: truncate 후 이어 쓰기
cw.writeChunk(List.of("chunk3-a", "chunk3-b"));
}
System.out.println("재시작 후 줄 수: " + Files.readAllLines(out).size()); // 8
System.out.println(Files.readAllLines(out).contains("partial-garbage")); // false
// 출력:
// 장애 후 줄 수: 7
// 재시작 후 줄 수: 8
// false