diff --git a/pom.xml b/pom.xml index 375cdf2..b58e98c 100644 --- a/pom.xml +++ b/pom.xml @@ -5,7 +5,7 @@ 4.0.0 com.codeforces.inmemo inmemo - 2.0.2-SNAPSHOT + 2.0.4-SNAPSHOT jar inmemo @@ -46,6 +46,7 @@ UTF-8 + -Dfile.encoding=UTF-8 -Xmx1200M @@ -76,16 +77,22 @@ maven-surefire-plugin 2.22.2 - -Dfile.encoding=UTF-8 - -Xmx1200M - - + ${surefire.argLine} + + + jdk9-plus-tests + + [9,) + + + -Dfile.encoding=UTF-8 -Xmx1200M --add-opens java.base/java.lang=ALL-UNNAMED --add-opens java.base/java.util=ALL-UNNAMED --add-opens java.base/java.io=ALL-UNNAMED + + + org.xerial.snappy diff --git a/src/main/java/com/codeforces/inmemo/JournalFormat.java b/src/main/java/com/codeforces/inmemo/JournalFormat.java new file mode 100644 index 0000000..e632f3a --- /dev/null +++ b/src/main/java/com/codeforces/inmemo/JournalFormat.java @@ -0,0 +1,90 @@ +package com.codeforces.inmemo; + +import org.apache.log4j.Logger; +import org.xerial.snappy.Snappy; + +import java.nio.charset.StandardCharsets; + +final class JournalFormat { + static final byte[] MAGIC = "INMEMO_JOURNAL_2".getBytes(StandardCharsets.US_ASCII); + static final int VERSION = 1; + + static final int MAX_BLOCK_ROWS = 1 << 20; + static final int MAX_BLOCK_RAW_BYTES = 128 * 1024 * 1024; + static final int MAX_BLOCK_COMPRESSED_BYTES = Snappy.maxCompressedLength(MAX_BLOCK_RAW_BYTES); + static final int MAX_HEADER_STRING_BYTES = 65_536; + + static final int DEFAULT_BLOCK_ROWS = 65_536; + static final int DEFAULT_TARGET_RAW_BYTES = 64 * 1024 * 1024; + static final int MIN_TARGET_RAW_BYTES = 1 * 1024 * 1024; + + static final String BLOCK_ROWS_PROPERTY = "Inmemo.JournalBlockRows"; + static final String TARGET_RAW_BYTES_PROPERTY = "Inmemo.JournalBlockTargetBytes"; + static final String MAX_AGE_HOURS_PROPERTY = "Inmemo.JournalMaxAgeHours"; + + private static final long DEFAULT_MAX_AGE_HOURS = 36L; + + private JournalFormat() { + // No operations. + } + + static int getBlockRows(Logger logger) { + return getIntProperty(logger, BLOCK_ROWS_PROPERTY, DEFAULT_BLOCK_ROWS, 1, MAX_BLOCK_ROWS); + } + + static int getTargetRawBytes(Logger logger) { + return getIntProperty(logger, TARGET_RAW_BYTES_PROPERTY, DEFAULT_TARGET_RAW_BYTES, + MIN_TARGET_RAW_BYTES, MAX_BLOCK_RAW_BYTES / 2); + } + + static long getMaxAgeMillis(Logger logger) { + long hours = getLongProperty(logger, MAX_AGE_HOURS_PROPERTY, DEFAULT_MAX_AGE_HOURS, 1L, 24L * 365L); + return hours * 60L * 60L * 1000L; + } + + private static int getIntProperty(Logger logger, String name, int defaultValue, int minValue, int maxValue) { + String value = System.getProperty(name); + if (value == null) { + return defaultValue; + } + + try { + int parsedValue = Integer.parseInt(value.trim()); + if (parsedValue < minValue) { + logger.warn("Property " + name + "=" + value + " is too small, using " + minValue + '.'); + return minValue; + } + if (parsedValue > maxValue) { + logger.warn("Property " + name + "=" + value + " is too large, using " + maxValue + '.'); + return maxValue; + } + return parsedValue; + } catch (NumberFormatException e) { + logger.warn("Property " + name + "=" + value + " is invalid, using " + defaultValue + '.', e); + return defaultValue; + } + } + + private static long getLongProperty(Logger logger, String name, long defaultValue, long minValue, long maxValue) { + String value = System.getProperty(name); + if (value == null) { + return defaultValue; + } + + try { + long parsedValue = Long.parseLong(value.trim()); + if (parsedValue < minValue) { + logger.warn("Property " + name + "=" + value + " is too small, using " + minValue + '.'); + return minValue; + } + if (parsedValue > maxValue) { + logger.warn("Property " + name + "=" + value + " is too large, using " + maxValue + '.'); + return maxValue; + } + return parsedValue; + } catch (NumberFormatException e) { + logger.warn("Property " + name + "=" + value + " is invalid, using " + defaultValue + '.', e); + return defaultValue; + } + } +} diff --git a/src/main/java/com/codeforces/inmemo/JournalReader.java b/src/main/java/com/codeforces/inmemo/JournalReader.java new file mode 100644 index 0000000..3a80d04 --- /dev/null +++ b/src/main/java/com/codeforces/inmemo/JournalReader.java @@ -0,0 +1,383 @@ +package com.codeforces.inmemo; + +import org.apache.log4j.Logger; +import org.jacuzzi.core.ArrayMap; +import org.jacuzzi.core.RowRoll; +import org.xerial.snappy.Snappy; + +import java.io.BufferedInputStream; +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.EOFException; +import java.io.File; +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; +import java.util.zip.CRC32; + +final class JournalReader implements AutoCloseable { + private static final Logger logger = Logger.getLogger(JournalReader.class); + private static final int STREAM_BUFFER_SIZE = 4 * 1024 * 1024; + private static final long MAX_FUTURE_CREATED_AT_MILLIS = 60L * 60L * 1000L; + + private final File file; + private final String expectedTableClassName; + private final String expectedTableClassSpec; + private final String tableNameForLog; + private final long fileLength; + + private CountingInputStream countingInputStream; + private DataInputStream inputStream; + private Status status; + private long blocksRead; + private long rowsRead; + private boolean closed; + + JournalReader(File file, Class tableClass, String tableClassSpec) { + this.file = file; + this.expectedTableClassName = ReflectionUtil.getTableClassName(tableClass); + this.expectedTableClassSpec = tableClassSpec; + this.tableNameForLog = tableClass.getSimpleName(); + this.fileLength = file.isFile() ? file.length() : 0L; + + if (!file.isFile()) { + status = Status.NO_FILE; + logger.info("Can't read journal because of no file '" + file + "'."); + return; + } + + try { + countingInputStream = new CountingInputStream(new BufferedInputStream( + Files.newInputStream(file.toPath()), STREAM_BUFFER_SIZE)); + inputStream = new DataInputStream(countingInputStream); + readHeader(); + } catch (EOFException e) { + status = Status.FORMAT_MISMATCH; + logger.warn("Journal header is truncated [table='" + tableNameForLog + "']."); + closeQuietly(); + } catch (Exception e) { + status = Status.FORMAT_MISMATCH; + logger.warn("Unable to read journal header [table='" + tableNameForLog + "'].", e); + closeQuietly(); + } + } + + Status getStatus() { + return status == null ? Status.CLEAN_EOF : status; + } + + long getRowsRead() { + return rowsRead; + } + + long getBlocksRead() { + return blocksRead; + } + + RowRoll nextBlock() { + if (status != null || closed) { + return null; + } + + if (countingInputStream.getPosition() == fileLength) { + finish(blocksRead == 0 ? Status.EMPTY : Status.CLEAN_EOF, null); + return null; + } + + try { + int rowCount = inputStream.readInt(); + int rawLength = inputStream.readInt(); + long expectedCrc32 = inputStream.readLong(); + int compressedLength = inputStream.readInt(); + + if (rowCount <= 0 || rowCount > JournalFormat.MAX_BLOCK_ROWS + || rawLength <= 0 || rawLength > JournalFormat.MAX_BLOCK_RAW_BYTES + || compressedLength <= 0 + || compressedLength > JournalFormat.MAX_BLOCK_COMPRESSED_BYTES + || compressedLength > fileLength - countingInputStream.getPosition() + || compressedLength > Snappy.maxCompressedLength(rawLength)) { + finish(Status.TRUNCATED, "Illegal journal block header"); + return null; + } + + byte[] compressed = new byte[compressedLength]; + inputStream.readFully(compressed); + + if (!Snappy.isValidCompressedBuffer(compressed)) { + finish(Status.TRUNCATED, "Invalid Snappy payload"); + return null; + } + + int uncompressedLength = Snappy.uncompressedLength(compressed); + if (uncompressedLength != rawLength) { + finish(Status.TRUNCATED, "Unexpected uncompressed journal block length"); + return null; + } + + byte[] raw = new byte[rawLength]; + int actualRawLength = Snappy.uncompress(compressed, 0, compressed.length, raw, 0); + if (actualRawLength != rawLength) { + finish(Status.TRUNCATED, "Unexpected actual uncompressed journal block length"); + return null; + } + + CRC32 crc32 = new CRC32(); + crc32.update(raw, 0, raw.length); + if (crc32.getValue() != expectedCrc32) { + finish(Status.TRUNCATED, "Journal block CRC mismatch"); + return null; + } + + ByteArrayInputStream rawInputStream = new ByteArrayInputStream(raw); + RowRoll rowRoll = ArrayMap.readRowRoll(rawInputStream); + if (rowRoll.size() != rowCount || rawInputStream.available() != 0) { + finish(Status.TRUNCATED, "Journal block payload mismatch"); + return null; + } + + blocksRead++; + rowsRead += rowRoll.size(); + return rowRoll; + } catch (EOFException e) { + finish(Status.TRUNCATED, "Unexpected journal EOF", e); + } catch (Exception e) { + finish(Status.TRUNCATED, "Unable to read journal block", e); + } + + return null; + } + + static RowRoll readAll(File file, Class tableClass, String tableClassSpec) { + try (JournalReader reader = new JournalReader(file, tableClass, tableClassSpec)) { + RowRoll result = null; + List resultKeys = null; + + RowRoll block; + while ((block = reader.nextBlock()) != null) { + if (block.isEmpty()) { + continue; + } + + List blockKeys = getKeys(block); + if (result == null) { + result = block; + resultKeys = blockKeys; + } else if (sameKeys(resultKeys, blockKeys)) { + result.add(block); + } else if (sameKeySet(resultKeys, blockKeys)) { + for (int i = 0; i < block.size(); i++) { + result.addRow(block.getRow(i)); + } + } else { + logger.warn("Journal blocks have different column sets [table='" + tableClass.getSimpleName() + "']."); + List unionKeys = unionKeys(resultKeys, blockKeys); + RowRoll rebuilt = null; + if (!sameKeys(resultKeys, unionKeys)) { + rebuilt = new RowRoll(); + rebuilt.setKeys(unionKeys.toArray(new String[0])); + for (int i = 0; i < result.size(); i++) { + rebuilt.addRow(result.getRow(i)); + } + } + for (int i = 0; i < block.size(); i++) { + if (rebuilt == null) { + result.addRow(block.getRow(i)); + } else { + rebuilt.addRow(block.getRow(i)); + } + } + if (rebuilt != null) { + result = rebuilt; + resultKeys = unionKeys; + } + } + } + + Status status = reader.getStatus(); + if (result == null) { + if (status == Status.NO_FILE) { + return null; + } + if (status == Status.EMPTY + || status == Status.FORMAT_MISMATCH + || status == Status.EXPIRED + || status == Status.TRUNCATED) { + return null; + } + } + + if (result != null) { + logger.info("Journal has been read [table='" + tableClass.getSimpleName() + + "', status=" + status + + ", blocks=" + reader.getBlocksRead() + + ", rows=" + reader.getRowsRead() + + ", bytes=" + file.length() + + "]."); + } + + return result; + } + } + + private void readHeader() throws IOException { + byte[] magic = new byte[JournalFormat.MAGIC.length]; + inputStream.readFully(magic); + if (!Arrays.equals(JournalFormat.MAGIC, magic)) { + finish(Status.FORMAT_MISMATCH, "Unexpected journal magic"); + return; + } + + int version = inputStream.readInt(); + if (version != JournalFormat.VERSION) { + finish(Status.FORMAT_MISMATCH, "Unexpected journal version"); + return; + } + + long createdAtMillis = inputStream.readLong(); + if (createdAtMillis <= 0) { + finish(Status.FORMAT_MISMATCH, "Illegal journal createdAtMillis"); + return; + } + + String tableClassName = readHeaderString(); + String tableClassSpec = readHeaderString(); + if (!expectedTableClassName.equals(tableClassName) || !expectedTableClassSpec.equals(tableClassSpec)) { + finish(Status.FORMAT_MISMATCH, "Journal table class mismatch"); + return; + } + + long nowMillis = System.currentTimeMillis(); + if (createdAtMillis > nowMillis + MAX_FUTURE_CREATED_AT_MILLIS + || nowMillis - createdAtMillis > JournalFormat.getMaxAgeMillis(logger)) { + expire(); + } + } + + private String readHeaderString() throws IOException { + int length = inputStream.readInt(); + if (length <= 0 || length > JournalFormat.MAX_HEADER_STRING_BYTES) { + throw new IOException("Illegal journal header string length: " + length + "."); + } + + byte[] bytes = new byte[length]; + inputStream.readFully(bytes); + return new String(bytes, StandardCharsets.UTF_8); + } + + private void expire() { + status = Status.EXPIRED; + closeQuietly(); + if (file.isFile() && !file.delete()) { + logger.warn("Expired journal has not been deleted [file='" + file.getAbsolutePath() + "']."); + } else { + logger.warn("Expired journal has been deleted [file='" + file.getAbsolutePath() + "']."); + } + } + + private void finish(Status status, String message) { + finish(status, message, null); + } + + private void finish(Status status, String message, Exception e) { + this.status = status; + closeQuietly(); + if (message != null) { + if (e == null) { + logger.warn(message + " [table='" + tableNameForLog + "', status=" + status + "]."); + } else { + logger.warn(message + " [table='" + tableNameForLog + "', status=" + status + "].", e); + } + } + } + + @Override + public void close() { + closeQuietly(); + } + + private void closeQuietly() { + if (closed) { + return; + } + + closed = true; + if (inputStream != null) { + try { + inputStream.close(); + } catch (IOException e) { + logger.error("Unable to close journal after reading [table='" + tableNameForLog + "'].", e); + } + } + } + + private static List getKeys(RowRoll rowRoll) { + return new ArrayList<>(rowRoll.getRow(0).keySet()); + } + + private static boolean sameKeys(List first, List second) { + return first.size() == second.size() && first.equals(second); + } + + private static boolean sameKeySet(List first, List second) { + return first.size() == second.size() && new LinkedHashSet<>(first).equals(new LinkedHashSet<>(second)); + } + + private static List unionKeys(List first, List second) { + Set keys = new LinkedHashSet<>(first); + keys.addAll(second); + return new ArrayList<>(keys); + } + + enum Status { + NO_FILE, + FORMAT_MISMATCH, + EXPIRED, + EMPTY, + CLEAN_EOF, + TRUNCATED + } + + private static final class CountingInputStream extends FilterInputStream { + private long position; + + private CountingInputStream(InputStream in) { + super(in); + } + + private long getPosition() { + return position; + } + + @Override + public int read() throws IOException { + int result = super.read(); + if (result != -1) { + position++; + } + return result; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + int result = super.read(b, off, len); + if (result > 0) { + position += result; + } + return result; + } + + @Override + public long skip(long n) throws IOException { + long result = super.skip(n); + position += result; + return result; + } + } +} diff --git a/src/main/java/com/codeforces/inmemo/JournalWriter.java b/src/main/java/com/codeforces/inmemo/JournalWriter.java new file mode 100644 index 0000000..7036b1e --- /dev/null +++ b/src/main/java/com/codeforces/inmemo/JournalWriter.java @@ -0,0 +1,350 @@ +package com.codeforces.inmemo; + +import org.apache.log4j.Logger; +import org.jacuzzi.core.ArrayMap; +import org.jacuzzi.core.Row; +import org.jacuzzi.core.RowRoll; +import org.xerial.snappy.Snappy; + +import java.io.BufferedOutputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.AtomicMoveNotSupportedException; +import java.nio.file.Files; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; +import java.util.Date; +import java.util.Map; +import java.util.zip.CRC32; + +final class JournalWriter { + private static final Logger logger = Logger.getLogger(JournalWriter.class); + private static final int STREAM_BUFFER_SIZE = 4 * 1024 * 1024; + + private final File file; + private final File tmpFile; + private final String tableClassName; + private final String tableClassSpec; + private final String tableNameForLog; + private final int blockRows; + private final int targetRawBytes; + private final boolean append; + + private DataOutputStream outputStream; + private RowRoll buffer = new RowRoll(); + private DirectByteArrayOutputStream rawOutputStream; + private byte[] compressedBuffer; + + private long estimatedRawBytes; + private long blocks; + private long rows; + private long maxRawBytes; + private long maxCompressedBytes; + private long maxFlushMillis; + private final long startTimeMillis = System.currentTimeMillis(); + + private boolean opened; + private boolean closed; + private boolean failed; + + JournalWriter(File file, Class tableClass, String tableClassSpec) { + this(file, tableClass, tableClassSpec, false); + } + + static JournalWriter append(File file, Class tableClass, String tableClassSpec) { + return new JournalWriter(file, tableClass, tableClassSpec, true); + } + + private JournalWriter(File file, Class tableClass, String tableClassSpec, boolean append) { + this.file = file; + this.tmpFile = new File(file.getParentFile(), file.getName() + ".tmp"); + this.tableClassName = ReflectionUtil.getTableClassName(tableClass); + this.tableClassSpec = tableClassSpec; + this.tableNameForLog = tableClass.getSimpleName(); + this.blockRows = JournalFormat.getBlockRows(logger); + this.targetRawBytes = JournalFormat.getTargetRawBytes(logger); + this.append = append; + } + + void addRow(Row row) { + if (closed || failed || row == null) { + return; + } + + try { + openIfNeeded(); + if (failed) { + return; + } + + buffer.addRow(row); + estimatedRawBytes += estimateRawBytes(row); + + if (buffer.size() >= blockRows || estimatedRawBytes >= targetRawBytes) { + flushBlock(); + } + } catch (Exception e) { + fail("Failed to add row to journal", e); + } + } + + void finish() { + if (closed) { + return; + } + + try { + if (!failed && buffer != null && !buffer.isEmpty()) { + flushBlock(); + } + + if (failed) { + closed = true; + return; + } + + if (!opened) { + closed = true; + return; + } + + closeOutputStreamQuietly(); + + if (append) { + closed = true; + logger.info("Journal has been appended [table='" + tableNameForLog + + "', blocks=" + blocks + + ", rows=" + rows + + ", maxRawBytes=" + maxRawBytes + + ", maxCompressedBytes=" + maxCompressedBytes + + ", maxFlushMillis=" + maxFlushMillis + + ", fileBytes=" + file.length() + + ", durationMillis=" + (System.currentTimeMillis() - startTimeMillis) + + "]."); + return; + } + + if (rows == 0) { + deleteTmpQuietly(); + closed = true; + return; + } + + moveTmpToTarget(); + closed = true; + + logger.info("Journal has been written [table='" + tableNameForLog + + "', blocks=" + blocks + + ", rows=" + rows + + ", maxRawBytes=" + maxRawBytes + + ", maxCompressedBytes=" + maxCompressedBytes + + ", maxFlushMillis=" + maxFlushMillis + + ", fileBytes=" + file.length() + + ", durationMillis=" + (System.currentTimeMillis() - startTimeMillis) + + "]."); + } catch (Exception e) { + fail("Failed to finish journal", e); + closed = true; + } + } + + private void openIfNeeded() throws IOException { + if (opened) { + return; + } + + if (append) { + if (!file.isFile()) { + fail("Can't append journal because file does not exist '" + file.getAbsolutePath() + "'", null); + return; + } + + outputStream = new DataOutputStream(new BufferedOutputStream(Files.newOutputStream(file.toPath(), + StandardOpenOption.WRITE, StandardOpenOption.APPEND), STREAM_BUFFER_SIZE)); + opened = true; + return; + } + + if (tmpFile.exists() && !tmpFile.delete()) { + fail("Can't delete stale temporary journal file '" + tmpFile.getAbsolutePath() + "'", null); + return; + } + + outputStream = new DataOutputStream(new BufferedOutputStream(Files.newOutputStream(tmpFile.toPath()), + STREAM_BUFFER_SIZE)); + writeHeader(); + opened = true; + } + + private void writeHeader() throws IOException { + outputStream.write(JournalFormat.MAGIC); + outputStream.writeInt(JournalFormat.VERSION); + outputStream.writeLong(System.currentTimeMillis()); + writeHeaderString(tableClassName); + writeHeaderString(tableClassSpec); + } + + private void writeHeaderString(String value) throws IOException { + byte[] bytes = value.getBytes(StandardCharsets.UTF_8); + if (bytes.length == 0 || bytes.length > JournalFormat.MAX_HEADER_STRING_BYTES) { + throw new IOException("Illegal journal header string length: " + bytes.length + "."); + } + outputStream.writeInt(bytes.length); + outputStream.write(bytes); + } + + private void flushBlock() throws IOException { + if (buffer == null || buffer.isEmpty()) { + return; + } + + long startTimeMillis = System.currentTimeMillis(); + int rowCount = buffer.size(); + + if (rawOutputStream == null) { + rawOutputStream = new DirectByteArrayOutputStream(estimateRawBufferSize()); + } + rawOutputStream.reset(); + rawOutputStream.ensureCapacity(estimateRawBufferSize()); + + ArrayMap.writeRowRoll(rawOutputStream, buffer); + int rawLength = rawOutputStream.size(); + if (rawLength <= 0 || rawLength > JournalFormat.MAX_BLOCK_RAW_BYTES) { + throw new IOException("Journal block raw length is out of bounds [table='" + tableNameForLog + + "', rawLength=" + rawLength + "]."); + } + + int maxCompressedLength = Snappy.maxCompressedLength(rawLength); + if (compressedBuffer == null || compressedBuffer.length < maxCompressedLength) { + compressedBuffer = new byte[maxCompressedLength]; + } + + byte[] rawBuffer = rawOutputStream.getBuffer(); + CRC32 crc32 = new CRC32(); + crc32.update(rawBuffer, 0, rawLength); + + int compressedLength = Snappy.compress(rawBuffer, 0, rawLength, compressedBuffer, 0); + if (compressedLength <= 0 || compressedLength > JournalFormat.MAX_BLOCK_COMPRESSED_BYTES) { + throw new IOException("Journal block compressed length is out of bounds [table='" + tableNameForLog + + "', compressedLength=" + compressedLength + "]."); + } + + outputStream.writeInt(rowCount); + outputStream.writeInt(rawLength); + outputStream.writeLong(crc32.getValue()); + outputStream.writeInt(compressedLength); + outputStream.write(compressedBuffer, 0, compressedLength); + + blocks++; + rows += rowCount; + maxRawBytes = Math.max(maxRawBytes, rawLength); + maxCompressedBytes = Math.max(maxCompressedBytes, compressedLength); + maxFlushMillis = Math.max(maxFlushMillis, System.currentTimeMillis() - startTimeMillis); + + buffer = new RowRoll(); + estimatedRawBytes = 0; + } + + private int estimateRawBufferSize() { + long size = estimatedRawBytes + estimatedRawBytes / 2 + 1024; + if (size < 1024) { + size = 1024; + } + if (size > JournalFormat.MAX_BLOCK_RAW_BYTES) { + size = JournalFormat.MAX_BLOCK_RAW_BYTES; + } + return (int) size; + } + + private long estimateRawBytes(Row row) { + long result = 16; + for (Map.Entry entry : row.entrySet()) { + // Keys are counted for every row on purpose: the estimate is an upper bound used for early flushing. + result += estimateString(entry.getKey()); + Object value = entry.getValue(); + if (value == null) { + result += 1; + } else if (value instanceof String) { + result += estimateString((String) value); + } else if (value instanceof Date + || value instanceof Byte + || value instanceof Integer + || value instanceof Long + || value instanceof Double + || value instanceof Boolean) { + result += 12; + } else { + result += 12; + } + } + return result; + } + + private long estimateString(String value) { + return value == null ? 1 : 4L + 2L * value.length(); + } + + private void moveTmpToTarget() throws IOException { + try { + Files.move(tmpFile.toPath(), file.toPath(), + StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + } catch (AtomicMoveNotSupportedException e) { + Files.move(tmpFile.toPath(), file.toPath(), StandardCopyOption.REPLACE_EXISTING); + } + } + + private void fail(String message, Exception e) { + failed = true; + closeOutputStreamQuietly(); + if (!append) { + deleteTmpQuietly(); + } + if (e == null) { + logger.error(message + " [table='" + tableNameForLog + "']."); + } else { + logger.error(message + " [table='" + tableNameForLog + "'].", e); + } + } + + private void closeOutputStreamQuietly() { + if (outputStream != null) { + try { + outputStream.close(); + } catch (IOException e) { + logger.error("Can't close journal output stream [table='" + tableNameForLog + "'].", e); + } finally { + outputStream = null; + } + } + } + + private void deleteTmpQuietly() { + if (tmpFile.isFile() && !tmpFile.delete()) { + logger.error("Can't delete temporary journal file '" + tmpFile.getAbsolutePath() + + "' [table='" + tableNameForLog + "']."); + } + } + + private static final class DirectByteArrayOutputStream extends ByteArrayOutputStream { + private DirectByteArrayOutputStream(int size) { + super(size); + } + + private byte[] getBuffer() { + return buf; + } + + private void ensureCapacity(int minCapacity) { + if (minCapacity <= buf.length) { + return; + } + + byte[] newBuffer = new byte[minCapacity]; + System.arraycopy(buf, 0, newBuffer, 0, count); + buf = newBuffer; + } + } +} diff --git a/src/main/java/com/codeforces/inmemo/Table.java b/src/main/java/com/codeforces/inmemo/Table.java index 5b9da16..59a4d8a 100644 --- a/src/main/java/com/codeforces/inmemo/Table.java +++ b/src/main/java/com/codeforces/inmemo/Table.java @@ -3,15 +3,14 @@ import gnu.trove.set.TLongSet; import gnu.trove.set.hash.TLongHashSet; import org.apache.log4j.Logger; -import org.jacuzzi.core.ArrayMap; import org.jacuzzi.core.Row; import org.jacuzzi.core.RowRoll; -import org.xerial.snappy.SnappyInputStream; -import org.xerial.snappy.SnappyOutputStream; import javax.annotation.Nonnull; import javax.annotation.Nullable; -import java.io.*; +import java.io.File; +import java.io.IOException; +import java.io.PrintWriter; import java.nio.file.Files; import java.util.ArrayList; import java.util.Date; @@ -31,7 +30,6 @@ public class Table { private static final Logger logger = Logger.getLogger(Table.class); private static final Pattern INDICATOR_FIELD_SPLIT_PATTERN = Pattern.compile("@"); - private static final int JOURNAL_STREAM_BUFFER_SIZE = 4 * 1024 * 1024; private final Lock lock = new ReentrantLock(); @@ -49,7 +47,7 @@ public class Table { private volatile boolean preloaded; private final TLongSet ids; - private RowRoll journal = new RowRoll(); + private JournalWriter journalWriter; private boolean useJournal = true; private static File journalsDir = new File("."); private final AtomicInteger insertOrUpdateCount = new AtomicInteger(); @@ -94,16 +92,44 @@ static void setJournalsDirForTesting(File journalsDir) { ids = Inmemo.getNoSizeSupportClasses().contains(clazz) ? null : new TLongHashSet(); this.rowFilter = rowFilter; if (Inmemo.isJournalSupportUnset(clazz)) { - journal = null; useJournal = false; deleteStaleJournalFileQuietly(); } + if (useJournal) { + journalWriter = createJournalWriter(); + } } void createUpdater(Object initialIndicatorValue) { + disableJournalWriter(); tableUpdater = new TableUpdater<>(this, initialIndicatorValue); } + void setJournalEligible(boolean journalEligible) { + if (!useJournal) { + setJournalWriter(null); + return; + } + + setJournalWriter(journalEligible ? createJournalWriter() : null); + } + + void disableJournalWriter() { + setJournalWriter(null); + } + + void startFreshJournalWriter() { + if (useJournal) { + setJournalWriter(createJournalWriter()); + } + } + + void startAppendJournalWriter() { + if (useJournal) { + setJournalWriter(JournalWriter.append(new File(journalsDir, getInmemoFilename()), clazz, clazzSpec)); + } + } + void runUpdater() { tableUpdater.start(); } @@ -227,8 +253,8 @@ int size() { private void internalInsertOrUpdate(@Nonnull T item, @Nullable Row row) { lock.lock(); try { - if (journal != null && row != null) { - journal.addRow(row); + if (journalWriter != null && row != null) { + journalWriter.addRow(row); } if (ids != null) { @@ -338,12 +364,20 @@ void logBucketStats() { void deleteJournal() throws IOException { File journalFile = new File(journalsDir, getInmemoFilename()); - if (journalFile.isFile()) { - if (!journalFile.delete()) { - String message = "Journal has not been deleted [table='" + clazz.getSimpleName() + "']."; - logger.error(message); - throw new IOException(message); - } + deleteFile(journalFile); + deleteFile(new File(journalsDir, getInmemoFilename() + ".tmp")); + } + + private void deleteFile(File file) throws IOException { + if (!file.isFile()) { + return; + } + + if (!file.delete()) { + String message = "Journal has not been deleted [table='" + clazz.getSimpleName() + + "', file='" + file.getAbsolutePath() + "']."; + logger.error(message); + throw new IOException(message); } } @@ -363,28 +397,18 @@ private String getInmemoFilename() { } void writeJournal() throws IOException { - if (useJournal && journal != null) { - if (journal.isEmpty()) { - logger.warn("Journal has not been dumped because of empty journal [table='" + clazz.getSimpleName() + "']."); - return; - } + if (!useJournal) { + return; + } - File journalFile = new File(journalsDir, getInmemoFilename()); - lock.lock(); - try { - long startTimeMillis = System.currentTimeMillis(); - try (OutputStream outputStream = new SnappyOutputStream(new BufferedOutputStream( - Files.newOutputStream(journalFile.toPath()), JOURNAL_STREAM_BUFFER_SIZE))) { - ArrayMap.writeRowRoll(outputStream, journal); - } - long writeRowRollTimeMillis = System.currentTimeMillis() - startTimeMillis; - logger.info("Journal binary data has been prepared and written in " - + writeRowRollTimeMillis + " ms [table='" + clazz.getSimpleName() - + "', size=" + journal.size() + ", bytes=" + journalFile.length() + "]."); - journal = null; - } finally { - lock.unlock(); + lock.lock(); + try { + if (journalWriter != null) { + journalWriter.finish(); + journalWriter = null; } + } finally { + lock.unlock(); } } @@ -394,39 +418,20 @@ RowRoll readJournal() { } File journalFile = new File(journalsDir, getInmemoFilename()); - if (journalFile.isFile()) { - long startTimeMillis = System.currentTimeMillis(); - InputStream inputStream = null; - try { - inputStream = new SnappyInputStream(new BufferedInputStream( - Files.newInputStream(journalFile.toPath()), JOURNAL_STREAM_BUFFER_SIZE)); - RowRoll rowRoll = ArrayMap.readRowRoll(inputStream); - long durationTimeMillis = System.currentTimeMillis() - startTimeMillis; - logger.info("Journal binary data has been read and parsed in " - + durationTimeMillis + " ms [table='" + clazz.getSimpleName() - + "', bytes=" + journalFile.length() + "]."); - return rowRoll; - } catch (ClassCastException e) { - logger.error("ClassCastException: Unable to read journal [table='" + clazz.getSimpleName() + "'].", e); - return null; - } catch (IOException e) { - logger.error("IOException: Unable to read journal [table='" + clazz.getSimpleName() + "'].", e); - return null; - } finally { - if (inputStream != null) { - try { - inputStream.close(); - } catch (IOException e) { - logger.error("Unable to close journal after reading [table='" + clazz.getSimpleName() + "'].", e); - // No operations. - } - } - } - } else { - logger.info("Can't read journal because of no file '" + journalFile + "'."); + try { + return JournalReader.readAll(journalFile, clazz, clazzSpec); + } catch (Exception e) { + logger.error("Unexpected exception while reading journal [table='" + clazz.getSimpleName() + "'].", e); + return null; } + } - return null; + JournalReader openJournalReader() { + return new JournalReader(new File(journalsDir, getInmemoFilename()), clazz, clazzSpec); + } + + TableUpdater getTableUpdaterForTesting() { + return tableUpdater; } boolean isUseJournal() { @@ -435,8 +440,29 @@ boolean isUseJournal() { /* init. */ { if (!"true".equals(System.getProperty("Inmemo.UseJournal"))) { - journal = null; useJournal = false; } } + + private JournalWriter createJournalWriter() { + return new JournalWriter(new File(journalsDir, getInmemoFilename()), clazz, clazzSpec); + } + + private void setJournalWriter(JournalWriter journalWriter) { + lock.lock(); + try { + if (this.journalWriter == journalWriter) { + return; + } + + JournalWriter oldJournalWriter = this.journalWriter; + this.journalWriter = null; + if (oldJournalWriter != null) { + oldJournalWriter.finish(); + } + this.journalWriter = journalWriter; + } finally { + lock.unlock(); + } + } } diff --git a/src/main/java/com/codeforces/inmemo/TableUpdater.java b/src/main/java/com/codeforces/inmemo/TableUpdater.java index e41e0db..d2a2cbc 100644 --- a/src/main/java/com/codeforces/inmemo/TableUpdater.java +++ b/src/main/java/com/codeforces/inmemo/TableUpdater.java @@ -57,6 +57,10 @@ class TableUpdater { private final long startTimeMillis; private final Map lastEntityIdsUpdateCount = new ConcurrentHashMap<>(); + private final boolean journalReplayEligible; + private boolean replayFinished; + private JournalReader replayReader; + private long replayedRows; TableUpdater(Table table, Object initialIndicatorValue) { if (dataSource == null) { @@ -68,6 +72,8 @@ class TableUpdater { this.table = table; this.lastIndicatorValue.set(initialIndicatorValue); + this.journalReplayEligible = initialIndicatorValue == null && table.isUseJournal(); + this.replayFinished = !journalReplayEligible; DataSource clazzDataSource = dataSourceByClazzName.get(table.getClazz().getName()); jacuzzi = Jacuzzi.getJacuzzi(clazzDataSource == null ? dataSource : clazzDataSource); @@ -218,18 +224,21 @@ void update() { } } - private void randomizedUpdate() { - List updatedIds = internalUpdate(); - sleepBetweenRescans(updatedIds.size()); + void randomizedUpdate() { + UpdateResult updateResult = internalUpdate(); + if (!updateResult.journalReplayInProgress) { + sleepBetweenRescans(updateResult.updatedIds.size()); + } } - private List internalUpdate() { + UpdateResult internalUpdate() { updateLock.lock(); try { long startTimeMillis = System.currentTimeMillis(); Object prevLastIndicatorValue = lastIndicatorValue.get(); - RowRoll rows = getRecentlyChangedRows(prevLastIndicatorValue); + RowsResult rowsResult = getRecentlyChangedRows(prevLastIndicatorValue); + RowRoll rows = rowsResult.rows; long afterGetRecentlyChangedRowsMillis = System.currentTimeMillis(); long getRecentlyChangedMillis = afterGetRecentlyChangedRowsMillis - startTimeMillis; @@ -337,7 +346,7 @@ private List internalUpdate() { + "]."); } - if (updatedIds.isEmpty() && !table.isPreloaded()) { + if (updatedIds.isEmpty() && !table.isPreloaded() && !rowsResult.journalReplayInProgress) { if (table.hasSize()) { logger.info("Inmemo ready to dump journal of table " + ReflectionUtil.getTableClassName(table.getClazz()) + " [items=" + table.size() + "]."); @@ -381,7 +390,7 @@ private List internalUpdate() { } } - return trulyUpdatedIds; + return new UpdateResult(trulyUpdatedIds, rowsResult.journalReplayInProgress); } finally { updateLock.unlock(); } @@ -414,7 +423,7 @@ private long getRescanTimeMillis() { return RESCAN_TIME_MILLIS; } - private void sleep(long timeMillis) { + void sleep(long timeMillis) { try { Thread.sleep(timeMillis); } catch (InterruptedException e) { @@ -423,25 +432,30 @@ private void sleep(long timeMillis) { } } - private RowRoll getRecentlyChangedRows(Object indicatorLastValue) { + private RowsResult getRecentlyChangedRows(Object indicatorLastValue) { long startTimeMillis = System.currentTimeMillis(); - RowRoll rows = null; - if (lastIndicatorValue.get() == null && table.isUseJournal()) { - RowRoll journalRows = table.readJournal(); + if (journalReplayEligible && !replayFinished) { + if (replayReader == null) { + replayReader = table.openJournalReader(); + } + + RowRoll journalRows = replayReader.nextBlock(); if (journalRows != null && !journalRows.isEmpty()) { - rows = journalRows; + replayedRows += journalRows.size(); + logger.info("getRecentlyChangedRows loads data of using the journal in " + + (System.currentTimeMillis() - startTimeMillis) + + " ms [table=" + table.getClazz().getSimpleName() + + ", blockRows=" + journalRows.size() + + ", replayedRows=" + replayedRows + "]."); + return new RowsResult(journalRows, true); } - } - if (rows != null) { - logger.info("getRecentlyChangedRows loads data of using the journal in " - + (System.currentTimeMillis() - startTimeMillis) - + " ms [table=" + table.getClazz().getSimpleName() + "]."); - return rows; + finishReplayAndConfigureWriter(replayReader.getStatus()); } String forceIndexClause = table.getDatabaseIndex() == null ? "" : ("FORCE INDEX (" + table.getDatabaseIndex() + ')'); + RowRoll rows; if (indicatorLastValue == null) { rows = jacuzzi.findRowRoll("SELECT * FROM " @@ -479,7 +493,57 @@ private RowRoll getRecentlyChangedRows(Object indicatorLastValue) { + queryTimeMillis + " ms."); } - return rows; + return new RowsResult(rows, false); + } + + private void finishReplayAndConfigureWriter(JournalReader.Status status) { + if (replayReader != null) { + replayReader.close(); + replayReader = null; + } + replayFinished = true; + + switch (status) { + case NO_FILE: + case FORMAT_MISMATCH: + case EXPIRED: + case EMPTY: + table.startFreshJournalWriter(); + logger.info("Journal replay finished, fresh writer has been enabled [table='" + + table.getClazz().getSimpleName() + + "', status=" + status + + ", replayedRows=" + replayedRows + "]."); + break; + case CLEAN_EOF: + table.startAppendJournalWriter(); + logger.info("Journal replay finished, append writer has been enabled [table='" + + table.getClazz().getSimpleName() + + "', status=" + status + + ", replayedRows=" + replayedRows + "]."); + break; + case TRUNCATED: + if (replayedRows == 0) { + table.startFreshJournalWriter(); + logger.warn("Journal replay finished on first corrupted block, fresh writer has been enabled [table='" + + table.getClazz().getSimpleName() + + "', status=" + status + + ", replayedRows=" + replayedRows + "]."); + } else { + table.disableJournalWriter(); + logger.warn("Journal replay finished with truncated tail, journal writer stays disabled [table='" + + table.getClazz().getSimpleName() + + "', status=" + status + + ", replayedRows=" + replayedRows + "]."); + } + break; + default: + table.disableJournalWriter(); + logger.warn("Journal replay finished with unexpected status, journal writer stays disabled [table='" + + table.getClazz().getSimpleName() + + "', status=" + status + + ", replayedRows=" + replayedRows + "]."); + break; + } } @SuppressWarnings("WeakerAccess") @@ -519,6 +583,26 @@ public void run() { } } + static final class UpdateResult { + final List updatedIds; + final boolean journalReplayInProgress; + + private UpdateResult(List updatedIds, boolean journalReplayInProgress) { + this.updatedIds = updatedIds; + this.journalReplayInProgress = journalReplayInProgress; + } + } + + private static final class RowsResult { + private final RowRoll rows; + private final boolean journalReplayInProgress; + + private RowsResult(RowRoll rows, boolean journalReplayInProgress) { + this.rows = rows; + this.journalReplayInProgress = journalReplayInProgress; + } + } + private static final class TableUpdaterThreadsPrinterRunnable implements Runnable { @SuppressWarnings("BusyWait") @Override diff --git a/src/test/java/com/codeforces/inmemo/JournalV2Test.java b/src/test/java/com/codeforces/inmemo/JournalV2Test.java new file mode 100644 index 0000000..d47e943 --- /dev/null +++ b/src/test/java/com/codeforces/inmemo/JournalV2Test.java @@ -0,0 +1,873 @@ +package com.codeforces.inmemo; + +import com.codeforces.inmemo.model.JournalDisabledUser; +import com.codeforces.inmemo.model.JournalEnabledUser; +import org.apache.log4j.Logger; +import org.hsqldb.jdbc.JDBCDataSource; +import org.jacuzzi.core.ArrayMap; +import org.jacuzzi.core.Row; +import org.jacuzzi.core.RowRoll; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.xerial.snappy.Snappy; +import org.xerial.snappy.SnappyInputStream; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.file.Files; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.util.Arrays; +import java.util.zip.CRC32; + +public class JournalV2Test { + private static final Logger logger = Logger.getLogger(JournalV2Test.class); + + private String oldUseJournalProperty; + private String oldBlockRowsProperty; + private String oldTargetBytesProperty; + private String oldMaxAgeProperty; + private File oldJournalsDir; + private File tempJournalsDir; + + @Before + public void setUp() throws Exception { + oldUseJournalProperty = System.getProperty("Inmemo.UseJournal"); + oldBlockRowsProperty = System.getProperty(JournalFormat.BLOCK_ROWS_PROPERTY); + oldTargetBytesProperty = System.getProperty(JournalFormat.TARGET_RAW_BYTES_PROPERTY); + oldMaxAgeProperty = System.getProperty(JournalFormat.MAX_AGE_HOURS_PROPERTY); + + System.setProperty("Inmemo.UseJournal", "true"); + System.setProperty(JournalFormat.BLOCK_ROWS_PROPERTY, "2"); + System.clearProperty(JournalFormat.TARGET_RAW_BYTES_PROPERTY); + System.clearProperty(JournalFormat.MAX_AGE_HOURS_PROPERTY); + + oldJournalsDir = Table.getJournalsDirForTesting(); + tempJournalsDir = Files.createTempDirectory("inmemo-journals-v2-").toFile(); + Table.setJournalsDir(tempJournalsDir); + } + + @After + public void tearDown() { + restoreProperty("Inmemo.UseJournal", oldUseJournalProperty); + restoreProperty(JournalFormat.BLOCK_ROWS_PROPERTY, oldBlockRowsProperty); + restoreProperty(JournalFormat.TARGET_RAW_BYTES_PROPERTY, oldTargetBytesProperty); + restoreProperty(JournalFormat.MAX_AGE_HOURS_PROPERTY, oldMaxAgeProperty); + + if (oldJournalsDir != null) { + Table.setJournalsDirForTesting(oldJournalsDir); + } + + deleteRecursively(tempJournalsDir); + } + + @Test + public void testRoundTripMultiblockAndDeletesStaleTmp() throws Exception { + File tmpFile = tmpJournalFile(JournalEnabledUser.class); + Files.write(tmpFile.toPath(), new byte[]{1, 2, 3}); + + Table table = new Table<>(JournalEnabledUser.class, "id", null); + for (long id = 1; id <= 5; id++) { + table.insertOrUpdate(user(id, "u" + id, "u" + id + "@example.com"), + row(id, "u" + id, "u" + id + "@example.com")); + } + table.writeJournal(); + + Assert.assertFalse(tmpFile.exists()); + Assert.assertTrue(journalFile(JournalEnabledUser.class).isFile()); + + RowRoll rows = table.readJournal(); + Assert.assertNotNull(rows); + Assert.assertEquals(5, rows.size()); + Assert.assertEquals("u5", rows.getRow(4).get("handle")); + + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + int blocks = 0; + while (reader.nextBlock() != null) { + blocks++; + } + Assert.assertTrue(blocks > 1); + Assert.assertEquals(JournalReader.Status.CLEAN_EOF, reader.getStatus()); + } + } + + @Test + public void testEmptyFinishClosesWriter() throws Exception { + Table table = new Table<>(JournalEnabledUser.class, "id", null); + table.writeJournal(); + + Assert.assertFalse(journalFile(JournalEnabledUser.class).exists()); + + table.insertOrUpdate(user(1L, "late", "late@example.com"), row(1L, "late", "late@example.com")); + table.writeJournal(); + + Assert.assertFalse(journalFile(JournalEnabledUser.class).exists()); + } + + @Test + public void testNonEligibleTableDoesNotWriteJournal() throws Exception { + Table table = new Table<>(JournalEnabledUser.class, "id", null); + table.setJournalEligible(false); + + table.insertOrUpdate(user(1L, "delta", "delta@example.com"), row(1L, "delta", "delta@example.com")); + table.writeJournal(); + + Assert.assertFalse(journalFile(JournalEnabledUser.class).exists()); + } + + @Test + public void testTruncatedTailKeepsWholePrefix() throws Exception { + writeFiveRows(); + + try (RandomAccessFile file = new RandomAccessFile(journalFile(JournalEnabledUser.class), "rw")) { + file.setLength(file.length() - 1); + } + + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + int rows = 0; + RowRoll block; + while ((block = reader.nextBlock()) != null) { + rows += block.size(); + } + + Assert.assertEquals(4, rows); + Assert.assertEquals(JournalReader.Status.TRUNCATED, reader.getStatus()); + } + } + + @Test + public void testCorruptedPayloadKeepsWholePrefix() throws Exception { + writeFiveRows(); + + File file = journalFile(JournalEnabledUser.class); + try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw")) { + randomAccessFile.seek(file.length() - 1); + int value = randomAccessFile.read(); + randomAccessFile.seek(file.length() - 1); + randomAccessFile.write(value ^ 1); + } + + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + int rows = 0; + RowRoll block; + while ((block = reader.nextBlock()) != null) { + rows += block.size(); + } + + Assert.assertEquals(4, rows); + Assert.assertEquals(JournalReader.Status.TRUNCATED, reader.getStatus()); + } + } + + @Test + public void testForeignMagicIsFormatMismatch() throws Exception { + Files.write(journalFile(JournalEnabledUser.class).toPath(), new byte[]{1, 2, 3}); + + Assert.assertNull(new Table<>(JournalEnabledUser.class, "id", null).readJournal()); + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + Assert.assertEquals(JournalReader.Status.FORMAT_MISMATCH, reader.getStatus()); + } + } + + @Test + public void testClassSpecMismatchIsFormatMismatch() throws Exception { + writeFiveRows(); + + try (JournalReader reader = new JournalReader(journalFile(JournalEnabledUser.class), + JournalDisabledUser.class, ReflectionUtil.getTableClassSpec(JournalDisabledUser.class))) { + Assert.assertEquals(JournalReader.Status.FORMAT_MISMATCH, reader.getStatus()); + } + } + + @Test + public void testExpiredJournalIsDeleted() throws Exception { + writeHeaderOnlyJournal(System.currentTimeMillis() - 48L * 60L * 60L * 1000L); + + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + Assert.assertEquals(JournalReader.Status.EXPIRED, reader.getStatus()); + } + + Assert.assertFalse(journalFile(JournalEnabledUser.class).exists()); + } + + @Test + public void testSchemaDriftReadAllUsesUnion() { + System.setProperty(JournalFormat.BLOCK_ROWS_PROPERTY, "1"); + + JournalWriter writer = new JournalWriter(journalFile(JournalEnabledUser.class), + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + writer.addRow(rowWithoutEmail(1L, "first")); + writer.addRow(row(2L, "second", "second@example.com")); + writer.finish(); + + RowRoll rows = JournalReader.readAll(journalFile(JournalEnabledUser.class), + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + + Assert.assertNotNull(rows); + Assert.assertEquals(2, rows.size()); + Assert.assertNull(rows.getRow(0).get("email")); + Assert.assertEquals("second@example.com", rows.getRow(1).get("email")); + } + + @Test + public void testSchemaDriftReadAllKeepsExistingUnionForSubsetBlocks() { + System.setProperty(JournalFormat.BLOCK_ROWS_PROPERTY, "1"); + + JournalWriter writer = new JournalWriter(journalFile(JournalEnabledUser.class), + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + writer.addRow(row(1L, "first", "first@example.com")); + writer.addRow(rowWithoutEmail(2L, "second")); + writer.addRow(rowWithoutEmail(3L, "third")); + writer.finish(); + + RowRoll rows = JournalReader.readAll(journalFile(JournalEnabledUser.class), + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + + Assert.assertNotNull(rows); + Assert.assertEquals(3, rows.size()); + Assert.assertEquals("first@example.com", rows.getRow(0).get("email")); + Assert.assertNull(rows.getRow(1).get("email")); + Assert.assertNull(rows.getRow(2).get("email")); + } + + @Test + public void testSameKeySetDifferentOrderReadAll() { + System.setProperty(JournalFormat.BLOCK_ROWS_PROPERTY, "1"); + + JournalWriter writer = new JournalWriter(journalFile(JournalEnabledUser.class), + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + writer.addRow(row(1L, "first", "first@example.com")); + writer.addRow(rowDifferentOrder(2L, "second", "second@example.com")); + writer.finish(); + + RowRoll rows = JournalReader.readAll(journalFile(JournalEnabledUser.class), + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + + Assert.assertNotNull(rows); + Assert.assertEquals(2, rows.size()); + Assert.assertEquals("first", rows.getRow(0).get("handle")); + Assert.assertEquals("second@example.com", rows.getRow(1).get("email")); + } + + @Test + public void testIllegalBlockHeadersDoNotAllocatePayload() throws Exception { + assertIllegalBlockHeader(Integer.MAX_VALUE, 1, 1); + assertIllegalBlockHeader(-1, 1, 1); + assertIllegalBlockHeader(1, Integer.MAX_VALUE, 1); + assertIllegalBlockHeader(1, -1, 1); + assertIllegalBlockHeader(1, 1, Integer.MAX_VALUE); + assertIllegalBlockHeader(1, 1, -1); + assertIllegalBlockHeader(1, 16, 1000); + + int compressedLength = Snappy.maxCompressedLength(1) + 1; + writeBlockHeaderOnly(1, 1, compressedLength); + try (RandomAccessFile file = new RandomAccessFile(journalFile(JournalEnabledUser.class), "rw")) { + file.setLength(file.length() + compressedLength); + } + assertReaderNextBlockStatus(JournalReader.Status.TRUNCATED); + } + + @Test + public void testIllegalHeaderStringLengthsAreFormatMismatch() throws Exception { + writeHeaderWithIllegalClassNameLength(Integer.MAX_VALUE); + assertReaderStatus(JournalReader.Status.FORMAT_MISMATCH); + + writeHeaderWithIllegalClassNameLength(0); + assertReaderStatus(JournalReader.Status.FORMAT_MISMATCH); + + writeHeaderWithIllegalClassSpecLength(Integer.MAX_VALUE); + assertReaderStatus(JournalReader.Status.FORMAT_MISMATCH); + + writeHeaderWithIllegalClassSpecLength(0); + assertReaderStatus(JournalReader.Status.FORMAT_MISMATCH); + } + + @Test + public void testBlockPayloadRowCountMismatchIsTruncated() throws Exception { + writeUpperRows(1L); + overwriteFirstBlockRowCount(journalFile(JournalEnabledUser.class), 2); + + assertReaderNextBlockStatus(JournalReader.Status.TRUNCATED); + } + + @Test + public void testBlockPayloadTrailingBytesAreTruncated() throws Exception { + RowRoll rowRoll = new RowRoll(); + rowRoll.addRow(row(1L, "first", "first@example.com")); + writePayloadBlock(rowRoll, new byte[]{0}); + + assertReaderNextBlockStatus(JournalReader.Status.TRUNCATED); + } + + @Test + public void testJournalCreatedTooFarInFutureIsExpired() throws Exception { + writeHeaderOnlyJournal(System.currentTimeMillis() + 2L * 60L * 60L * 1000L); + + assertReaderStatus(JournalReader.Status.EXPIRED); + Assert.assertFalse(journalFile(JournalEnabledUser.class).exists()); + } + + @Test + public void testNonPositiveJournalCreatedAtIsFormatMismatch() throws Exception { + writeHeaderOnlyJournal(0L); + assertReaderStatus(JournalReader.Status.FORMAT_MISMATCH); + Assert.assertTrue(journalFile(JournalEnabledUser.class).exists()); + + writeHeaderOnlyJournal(-1L); + assertReaderStatus(JournalReader.Status.FORMAT_MISMATCH); + Assert.assertTrue(journalFile(JournalEnabledUser.class).exists()); + } + + @Test + public void testJournalPropertiesAreClampedAndDefaulted() { + System.setProperty(JournalFormat.BLOCK_ROWS_PROPERTY, "0"); + Assert.assertEquals(1, JournalFormat.getBlockRows(logger)); + System.setProperty(JournalFormat.BLOCK_ROWS_PROPERTY, Integer.toString(Integer.MAX_VALUE)); + Assert.assertEquals(JournalFormat.MAX_BLOCK_ROWS, JournalFormat.getBlockRows(logger)); + System.setProperty(JournalFormat.BLOCK_ROWS_PROPERTY, "bad"); + Assert.assertEquals(JournalFormat.DEFAULT_BLOCK_ROWS, JournalFormat.getBlockRows(logger)); + + System.setProperty(JournalFormat.TARGET_RAW_BYTES_PROPERTY, "1"); + Assert.assertEquals(JournalFormat.MIN_TARGET_RAW_BYTES, JournalFormat.getTargetRawBytes(logger)); + System.setProperty(JournalFormat.TARGET_RAW_BYTES_PROPERTY, + Integer.toString(JournalFormat.MAX_BLOCK_RAW_BYTES)); + Assert.assertEquals(JournalFormat.MAX_BLOCK_RAW_BYTES / 2, JournalFormat.getTargetRawBytes(logger)); + System.setProperty(JournalFormat.TARGET_RAW_BYTES_PROPERTY, "bad"); + Assert.assertEquals(JournalFormat.DEFAULT_TARGET_RAW_BYTES, JournalFormat.getTargetRawBytes(logger)); + + System.setProperty(JournalFormat.MAX_AGE_HOURS_PROPERTY, "0"); + Assert.assertEquals(60L * 60L * 1000L, JournalFormat.getMaxAgeMillis(logger)); + System.setProperty(JournalFormat.MAX_AGE_HOURS_PROPERTY, Long.toString(24L * 365L + 1L)); + Assert.assertEquals(24L * 365L * 60L * 60L * 1000L, JournalFormat.getMaxAgeMillis(logger)); + System.setProperty(JournalFormat.MAX_AGE_HOURS_PROPERTY, "bad"); + Assert.assertEquals(36L * 60L * 60L * 1000L, JournalFormat.getMaxAgeMillis(logger)); + } + + @Test + public void testUnsupportedValueDisablesWriterWithoutException() throws Exception { + Table table = new Table<>(JournalEnabledUser.class, "id", null); + Row row = new Row(4); + row.put("id", 1L); + row.put("handle", "bad"); + row.put("email", "bad@example.com"); + row.put("payload", new byte[]{1, 2, 3}); + + table.insertOrUpdate(user(1L, "bad", "bad@example.com"), row); + table.writeJournal(); + + Assert.assertFalse(journalFile(JournalEnabledUser.class).exists()); + Assert.assertFalse(tmpJournalFile(JournalEnabledUser.class).exists()); + } + + @Test + public void testReplayBlocksThenAppendSqlDelta() throws Exception { + createDataSourceWithRows(1L, 2L, 3L, 4L, 5L); + writeUpperRows(1L, 2L, 3L); + + File file = journalFile(JournalEnabledUser.class); + byte[] originalBytes = Files.readAllBytes(file.toPath()); + long originalCreatedAtMillis = readCreatedAtMillis(file); + + Table table = createUpdaterTable(); + TableUpdater updater = table.getTableUpdaterForTesting(); + + TableUpdater.UpdateResult firstBlock = updater.internalUpdate(); + Assert.assertTrue(firstBlock.journalReplayInProgress); + Assert.assertEquals(2, table.size()); + Assert.assertFalse(table.isPreloaded()); + + TableUpdater.UpdateResult secondBlock = updater.internalUpdate(); + Assert.assertTrue(secondBlock.journalReplayInProgress); + Assert.assertEquals(3, table.size()); + Assert.assertFalse(table.isPreloaded()); + + TableUpdater.UpdateResult sqlDelta = updater.internalUpdate(); + Assert.assertFalse(sqlDelta.journalReplayInProgress); + Assert.assertEquals(5, table.size()); + Assert.assertFalse(table.isPreloaded()); + + TableUpdater.UpdateResult finish = updater.internalUpdate(); + Assert.assertFalse(finish.journalReplayInProgress); + Assert.assertTrue(table.isPreloaded()); + + byte[] appendedBytes = Files.readAllBytes(file.toPath()); + Assert.assertTrue(appendedBytes.length > originalBytes.length); + Assert.assertTrue(startsWith(appendedBytes, originalBytes)); + Assert.assertEquals(originalCreatedAtMillis, readCreatedAtMillis(file)); + + RowRoll rows = JournalReader.readAll(file, + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + Assert.assertNotNull(rows); + Assert.assertEquals(5, rows.size()); + Assert.assertEquals(5L, rows.getRow(4).get("ID")); + } + + @Test + public void testRandomizedUpdateDoesNotSleepBetweenJournalBlocks() throws Exception { + createDataSourceWithRows(1L, 2L, 3L, 4L, 5L); + writeUpperRows(1L, 2L, 3L); + + Table table = new Table<>(JournalEnabledUser.class, "ID", null); + table.disableJournalWriter(); + RecordingTableUpdater updater = new RecordingTableUpdater(table); + + updater.randomizedUpdate(); + Assert.assertEquals(0, updater.sleepCount); + Assert.assertEquals(2, table.size()); + + updater.randomizedUpdate(); + Assert.assertEquals(0, updater.sleepCount); + Assert.assertEquals(3, table.size()); + + updater.randomizedUpdate(); + Assert.assertEquals(1, updater.sleepCount); + Assert.assertEquals(5, table.size()); + + table.writeJournal(); + } + + @Test + public void testUpdaterWithInitialIndicatorValueDoesNotWriteJournal() throws Exception { + createDataSourceWithRows(1L, 2L); + + Table table = new Table<>(JournalEnabledUser.class, "ID", null); + table.createUpdater(1L); + runUpdaterUntilPreloaded(table, 4); + + Assert.assertFalse(journalFile(JournalEnabledUser.class).exists()); + Assert.assertFalse(tmpJournalFile(JournalEnabledUser.class).exists()); + } + + @Test + public void testTruncatedTailWithPrefixDoesNotRewriteJournal() throws Exception { + createDataSourceWithRows(1L, 2L, 3L, 4L, 5L, 6L); + writeUpperRows(1L, 2L, 3L, 4L, 5L); + + File file = journalFile(JournalEnabledUser.class); + try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw")) { + randomAccessFile.setLength(randomAccessFile.length() - 1); + } + byte[] truncatedBytes = Files.readAllBytes(file.toPath()); + + Table table = createUpdaterTable(); + runUpdaterUntilPreloaded(table, 8); + + Assert.assertEquals(6, table.size()); + Assert.assertArrayEquals(truncatedBytes, Files.readAllBytes(file.toPath())); + + RowRoll rows = JournalReader.readAll(file, + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + Assert.assertNotNull(rows); + Assert.assertEquals(4, rows.size()); + } + + @Test + public void testTruncatedFirstBlockIsReplacedByFreshJournal() throws Exception { + createDataSourceWithRows(1L, 2L, 3L, 4L); + writeUpperRows(1L, 2L, 3L); + + File file = journalFile(JournalEnabledUser.class); + overwriteFirstBlockRowCount(file, 0); + byte[] corruptedBytes = Files.readAllBytes(file.toPath()); + + Table table = createUpdaterTable(); + runUpdaterUntilPreloaded(table, 6); + + byte[] healedBytes = Files.readAllBytes(file.toPath()); + Assert.assertFalse(Arrays.equals(corruptedBytes, healedBytes)); + + RowRoll rows = JournalReader.readAll(file, + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + Assert.assertNotNull(rows); + Assert.assertEquals(4, rows.size()); + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + while (reader.nextBlock() != null) { + // No operations. + } + Assert.assertEquals(JournalReader.Status.CLEAN_EOF, reader.getStatus()); + } + } + + @Test + public void testEmptyAppendFinishDoesNotTouchExistingJournal() throws Exception { + writeUpperRows(1L, 2L); + + File file = journalFile(JournalEnabledUser.class); + byte[] before = Files.readAllBytes(file.toPath()); + long lastModified = file.lastModified(); + + JournalWriter.append(file, + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)).finish(); + + Assert.assertArrayEquals(before, Files.readAllBytes(file.toPath())); + Assert.assertEquals(lastModified, file.lastModified()); + } + + @Test + public void testOpeningTerminalStatusesCreateFreshJournalBeforeSql() throws Exception { + assertTerminalStatusCreatesFreshJournal(this::deleteJournalFiles); + assertTerminalStatusCreatesFreshJournal(() -> Files.write(journalFile(JournalEnabledUser.class).toPath(), + new byte[]{1, 2, 3})); + assertTerminalStatusCreatesFreshJournal(() -> writeHeaderOnlyJournal(System.currentTimeMillis() + - 48L * 60L * 60L * 1000L)); + assertTerminalStatusCreatesFreshJournal(() -> writeHeaderOnlyJournal(System.currentTimeMillis())); + } + + @Test + public void testRollbackPreflightOldReaderGetsCatchableException() throws Exception { + writeFiveRows(); + + byte[] journalBytes = Files.readAllBytes(journalFile(JournalEnabledUser.class).toPath()); + try (SnappyInputStream inputStream = new SnappyInputStream(new ByteArrayInputStream(journalBytes))) { + ArrayMap.readRowRoll(inputStream); + Assert.fail("Old reader must not read v2 journal."); + } catch (IOException | ClassCastException expected) { + // Expected. + } + } + + private void writeFiveRows() throws Exception { + Table table = new Table<>(JournalEnabledUser.class, "id", null); + for (long id = 1; id <= 5; id++) { + table.insertOrUpdate(user(id, "u" + id, "u" + id + "@example.com"), + row(id, "u" + id, "u" + id + "@example.com")); + } + table.writeJournal(); + } + + private void writeUpperRows(long... ids) { + JournalWriter writer = new JournalWriter(journalFile(JournalEnabledUser.class), + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + for (long id : ids) { + writer.addRow(upperRow(id)); + } + writer.finish(); + } + + private void writeHeaderOnlyJournal(long createdAtMillis) throws IOException { + try (DataOutputStream outputStream = new DataOutputStream(Files.newOutputStream( + journalFile(JournalEnabledUser.class).toPath()))) { + writeJournalHeader(outputStream, createdAtMillis); + } + } + + private void writeHeaderWithIllegalClassNameLength(int classNameLength) throws IOException { + try (DataOutputStream outputStream = new DataOutputStream(Files.newOutputStream( + journalFile(JournalEnabledUser.class).toPath()))) { + outputStream.write(JournalFormat.MAGIC); + outputStream.writeInt(JournalFormat.VERSION); + outputStream.writeLong(System.currentTimeMillis()); + outputStream.writeInt(classNameLength); + } + } + + private void writeHeaderWithIllegalClassSpecLength(int classSpecLength) throws IOException { + try (DataOutputStream outputStream = new DataOutputStream(Files.newOutputStream( + journalFile(JournalEnabledUser.class).toPath()))) { + outputStream.write(JournalFormat.MAGIC); + outputStream.writeInt(JournalFormat.VERSION); + outputStream.writeLong(System.currentTimeMillis()); + writeHeaderString(outputStream, ReflectionUtil.getTableClassName(JournalEnabledUser.class)); + outputStream.writeInt(classSpecLength); + } + } + + private void assertIllegalBlockHeader(int rowCount, int rawLength, int compressedLength) throws IOException { + writeBlockHeaderOnly(rowCount, rawLength, compressedLength); + assertReaderNextBlockStatus(JournalReader.Status.TRUNCATED); + } + + private void writeBlockHeaderOnly(int rowCount, int rawLength, int compressedLength) throws IOException { + try (DataOutputStream outputStream = new DataOutputStream(Files.newOutputStream( + journalFile(JournalEnabledUser.class).toPath()))) { + writeJournalHeader(outputStream, System.currentTimeMillis()); + outputStream.writeInt(rowCount); + outputStream.writeInt(rawLength); + outputStream.writeLong(0L); + outputStream.writeInt(compressedLength); + } + } + + private void writePayloadBlock(RowRoll rowRoll, byte[] trailingBytes) throws IOException { + ByteArrayOutputStream rawOutputStream = new ByteArrayOutputStream(); + ArrayMap.writeRowRoll(rawOutputStream, rowRoll); + if (trailingBytes != null) { + rawOutputStream.write(trailingBytes); + } + + byte[] raw = rawOutputStream.toByteArray(); + byte[] compressed = new byte[Snappy.maxCompressedLength(raw.length)]; + int compressedLength = Snappy.compress(raw, 0, raw.length, compressed, 0); + + CRC32 crc32 = new CRC32(); + crc32.update(raw); + + try (DataOutputStream outputStream = new DataOutputStream(Files.newOutputStream( + journalFile(JournalEnabledUser.class).toPath()))) { + writeJournalHeader(outputStream, System.currentTimeMillis()); + outputStream.writeInt(rowRoll.size()); + outputStream.writeInt(raw.length); + outputStream.writeLong(crc32.getValue()); + outputStream.writeInt(compressedLength); + outputStream.write(compressed, 0, compressedLength); + } + } + + private void writeJournalHeader(DataOutputStream outputStream, long createdAtMillis) throws IOException { + outputStream.write(JournalFormat.MAGIC); + outputStream.writeInt(JournalFormat.VERSION); + outputStream.writeLong(createdAtMillis); + writeHeaderString(outputStream, ReflectionUtil.getTableClassName(JournalEnabledUser.class)); + writeHeaderString(outputStream, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + } + + private void createDataSourceWithRows(long... ids) throws Exception { + JDBCDataSource dataSource = new JDBCDataSource(); + dataSource.setUrl("jdbc:hsqldb:mem:journal-v2-" + System.nanoTime()); + dataSource.setUser("sa"); + dataSource.setPassword(""); + + try (Connection connection = dataSource.getConnection()) { + connection.createStatement().execute("DROP TABLE JournalEnabledUser IF EXISTS"); + connection.createStatement().execute("CREATE TABLE JournalEnabledUser (" + + "ID BIGINT, " + + "HANDLE VARCHAR(255), " + + "EMAIL VARCHAR(255))"); + + try (PreparedStatement statement = connection.prepareStatement( + "INSERT INTO JournalEnabledUser (ID, HANDLE, EMAIL) VALUES (?, ?, ?)")) { + for (long id : ids) { + statement.setLong(1, id); + statement.setString(2, "u" + id); + statement.setString(3, "u" + id + "@example.com"); + statement.executeUpdate(); + } + } + } + + TableUpdater.setDataSource(dataSource); + } + + private Table createUpdaterTable() { + Table table = new Table<>(JournalEnabledUser.class, "ID", null); + table.createUpdater(null); + return table; + } + + private void runUpdaterUntilPreloaded(Table table, int maxIterations) { + TableUpdater updater = table.getTableUpdaterForTesting(); + for (int i = 0; i < maxIterations; i++) { + updater.internalUpdate(); + if (table.isPreloaded()) { + return; + } + } + Assert.fail("Table has not been preloaded after " + maxIterations + " update iterations."); + } + + private void assertTerminalStatusCreatesFreshJournal(ThrowingRunnable fileSetup) throws Exception { + deleteJournalFiles(); + fileSetup.run(); + createDataSourceWithRows(1L, 2L); + + File file = journalFile(JournalEnabledUser.class); + Table table = createUpdaterTable(); + runUpdaterUntilPreloaded(table, 4); + + Assert.assertTrue(file.isFile()); + RowRoll rows = JournalReader.readAll(file, + JournalEnabledUser.class, ReflectionUtil.getTableClassSpec(JournalEnabledUser.class)); + Assert.assertNotNull(rows); + Assert.assertEquals(2, rows.size()); + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + while (reader.nextBlock() != null) { + // No operations. + } + Assert.assertEquals(JournalReader.Status.CLEAN_EOF, reader.getStatus()); + } + } + + private void deleteJournalFiles() throws IOException { + Files.deleteIfExists(journalFile(JournalEnabledUser.class).toPath()); + Files.deleteIfExists(tmpJournalFile(JournalEnabledUser.class).toPath()); + } + + private static boolean startsWith(byte[] bytes, byte[] prefix) { + if (bytes.length < prefix.length) { + return false; + } + + for (int i = 0; i < prefix.length; i++) { + if (bytes[i] != prefix[i]) { + return false; + } + } + + return true; + } + + private static long readCreatedAtMillis(File file) throws IOException { + try (DataInputStream inputStream = new DataInputStream(new BufferedInputStream( + Files.newInputStream(file.toPath())))) { + byte[] magic = new byte[JournalFormat.MAGIC.length]; + inputStream.readFully(magic); + Assert.assertArrayEquals(JournalFormat.MAGIC, magic); + Assert.assertEquals(JournalFormat.VERSION, inputStream.readInt()); + return inputStream.readLong(); + } + } + + private void assertReaderStatus(JournalReader.Status expectedStatus) { + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + Assert.assertEquals(expectedStatus, reader.getStatus()); + } + } + + private void assertReaderNextBlockStatus(JournalReader.Status expectedStatus) { + try (JournalReader reader = newReader(JournalEnabledUser.class)) { + Assert.assertNull(reader.nextBlock()); + Assert.assertEquals(expectedStatus, reader.getStatus()); + } + } + + private static void overwriteFirstBlockRowCount(File file, int rowCount) throws IOException { + try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw")) { + randomAccessFile.seek(getFirstBlockOffset(file)); + randomAccessFile.writeInt(rowCount); + } + } + + private static long getFirstBlockOffset(File file) throws IOException { + long offset = JournalFormat.MAGIC.length + 4L + 8L; + try (DataInputStream inputStream = new DataInputStream(new BufferedInputStream( + Files.newInputStream(file.toPath())))) { + skipFully(inputStream, JournalFormat.MAGIC.length + 4 + 8); + int tableClassNameLength = inputStream.readInt(); + offset += 4L + tableClassNameLength; + skipFully(inputStream, tableClassNameLength); + int tableClassSpecLength = inputStream.readInt(); + offset += 4L + tableClassSpecLength; + } + return offset; + } + + private static void skipFully(DataInputStream inputStream, int bytes) throws IOException { + int skipped = 0; + while (skipped < bytes) { + int current = inputStream.skipBytes(bytes - skipped); + if (current <= 0) { + throw new IOException("Unexpected EOF while skipping " + bytes + " bytes."); + } + skipped += current; + } + } + + private void writeHeaderString(DataOutputStream outputStream, String value) throws IOException { + byte[] bytes = value.getBytes("UTF-8"); + outputStream.writeInt(bytes.length); + outputStream.write(bytes); + } + + private JournalReader newReader(Class clazz) { + return new JournalReader(journalFile(clazz), clazz, ReflectionUtil.getTableClassSpec(clazz)); + } + + private static File journalFile(Class clazz) { + return new File(Table.getJournalsDirForTesting(), clazz.getSimpleName() + ".inmemo"); + } + + private static File tmpJournalFile(Class clazz) { + return new File(Table.getJournalsDirForTesting(), clazz.getSimpleName() + ".inmemo.tmp"); + } + + private static JournalEnabledUser user(long id, String handle, String email) { + JournalEnabledUser user = new JournalEnabledUser(); + user.setId(id); + user.setHandle(handle); + user.setEmail(email); + return user; + } + + private static Row row(long id, String handle, String email) { + Row row = new Row(3); + row.put("id", id); + row.put("handle", handle); + row.put("email", email); + return row; + } + + private static Row rowDifferentOrder(long id, String handle, String email) { + Row row = new Row(3); + row.put("email", email); + row.put("handle", handle); + row.put("id", id); + return row; + } + + private static Row upperRow(long id) { + Row row = new Row(3); + row.put("ID", id); + row.put("HANDLE", "u" + id); + row.put("EMAIL", "u" + id + "@example.com"); + return row; + } + + private static Row rowWithoutEmail(long id, String handle) { + Row row = new Row(2); + row.put("id", id); + row.put("handle", handle); + return row; + } + + private static void restoreProperty(String name, String value) { + if (value == null) { + System.clearProperty(name); + } else { + System.setProperty(name, value); + } + } + + private static void deleteRecursively(File file) { + if (file == null || !file.exists()) { + return; + } + + if (file.isDirectory()) { + File[] files = file.listFiles(); + if (files != null) { + for (File child : files) { + deleteRecursively(child); + } + } + } + + if (!file.delete() && file.exists()) { + throw new AssertionError("Can't delete " + file.getAbsolutePath()); + } + } + + private interface ThrowingRunnable { + void run() throws Exception; + } + + private static final class RecordingTableUpdater extends TableUpdater { + private int sleepCount; + + private RecordingTableUpdater(Table table) { + super(table, null); + } + + @Override + void sleep(long timeMillis) { + sleepCount++; + } + } +}