diff options
Diffstat (limited to 'src')
4 files changed, 45 insertions, 49 deletions
diff --git a/src/main/java/com/keuin/blame/SubmitWorker.java b/src/main/java/com/keuin/blame/SubmitWorker.java index 3c33fd6..ba4a922 100644 --- a/src/main/java/com/keuin/blame/SubmitWorker.java +++ b/src/main/java/com/keuin/blame/SubmitWorker.java @@ -13,9 +13,7 @@ import org.jetbrains.annotations.NotNull; import java.io.IOException; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; import static com.clickhouse.client.ClickHouseCredentials.fromUserAndPassword; @@ -90,27 +88,43 @@ public class SubmitWorker { private void bulkWrite( List<LogEntry> buffer, ClickHouseRequest.Mutation req - ) throws IOException, ClickHouseException { + ) throws IOException, ClickHouseException, ExecutionException, InterruptedException { + if (buffer.isEmpty()) { + logger.error("bulkWrite is called with empty write buffer"); + return; + } + logger.info("bulkWrite size: " + buffer.size()); + CompletableFuture<ClickHouseResponse> fut; try (var os = ClickHouseDataStreamFactory.getInstance() - .createPipedOutputStream(req.getConfig(), (Runnable) null); - var resp = req.data(os.getInputStream()).executeAndWait()) { + .createPipedOutputStream(req.getConfig())) { + fut = req.data(os.getInputStream()).execute(); for (var el : buffer) { el.write(os); } - logger.info(String.format("Written %d log entries to ClickHouse. %s", - buffer.size(), resp.getSummary().toString())); + } + try (var resp = fut.get()) { + var summary = resp.getSummary(); + var expected = buffer.size(); + var actual = summary.getReadRows(); + logger.info("Write success: " + summary.toString()); + if (expected != actual) { + logger.error(String.format( + "Unexpected write rows, expected %d (buffer), actual %d (write summary)", + expected, actual)); + } } buffer.clear(); } private void run() { var config = DatabaseUtil.DB_CONFIG; + logger.info("Database: " + config.toString()); var server = ClickHouseNode.builder() .host(config.address()) .port(ClickHouseProtocol.HTTP, config.port()) // .port(ClickHouseProtocol.GRPC, Integer.getInteger("chPort", 9100)) // .port(ClickHouseProtocol.TCP, Integer.getInteger("chPort", 9000)) - .database(config.table()) + .database(config.database()) .credentials(fromUserAndPassword(config.username(), config.password())) .build(); @@ -121,7 +135,12 @@ public class SubmitWorker { try (var client = ClickHouseClient.newInstance(server.getProtocol())) { writeLoop: while (true) { - var result = work(client, server, batchBuffer, writeImmediately); + var req = client.read(server).write() + .table(config.table()) + .format(ClickHouseFormat.RowBinary) +// .option(ClickHouseClientOption.ASYNC, false) + .option(ClickHouseClientOption.COMPRESS, false); + var result = work(client, req, batchBuffer, writeImmediately); switch (result) { case CONTINUE -> { writeImmediately = false; @@ -133,6 +152,9 @@ public class SubmitWorker { case FINISH -> { break workLoop; } + case INSTANT_WRITE -> { + writeImmediately = true; + } } } } @@ -148,12 +170,10 @@ public class SubmitWorker { private @NotNull WorkResult work( ClickHouseClient client, - ClickHouseNode server, + ClickHouseRequest.Mutation req, List<LogEntry> buffer, boolean writeImmediately ) { - var req = client.write(server).format(ClickHouseFormat.RowBinary) - .option(ClickHouseClientOption.ASYNC, false); boolean interrupted = false; // if writeImmediately is set, do not accumulate the buffer, flush it immediately try { @@ -177,13 +197,15 @@ public class SubmitWorker { return WorkResult.FINISH; } return WorkResult.RECONNECT; - } catch (ClickHouseException ex) { + } catch (ClickHouseException | ExecutionException ex) { logger.error("ClickHouse writer error", ex); try { Thread.sleep(3000); } catch (InterruptedException ignored) { } return WorkResult.RECONNECT; + } catch (InterruptedException e) { + return WorkResult.INSTANT_WRITE; } return WorkResult.CONTINUE; } diff --git a/src/main/java/com/keuin/blame/data/entry/LogEntry.java b/src/main/java/com/keuin/blame/data/entry/LogEntry.java index 839afdd..fba04be 100644 --- a/src/main/java/com/keuin/blame/data/entry/LogEntry.java +++ b/src/main/java/com/keuin/blame/data/entry/LogEntry.java @@ -10,7 +10,6 @@ import com.keuin.blame.util.UuidUtils; import net.minecraft.MinecraftVersion; import java.io.IOException; -import java.io.OutputStream; import java.util.Objects; import java.util.UUID; @@ -148,49 +147,34 @@ public class LogEntry { } public void write(ClickHousePipedOutputStream os) throws IOException { - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeInt8(os, actionType.getValue()); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeFixedString(os, gameVersion, 8); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeString(os, objectId); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeFixedString(os, objectPos.getWorld(), 24); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeInt64(os, (long) (objectPos.getX())); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeInt64(os, (long) (objectPos.getY())); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeInt64(os, (long) (objectPos.getZ())); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeInt32(os, objectType.getValue()); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeString(os, subjectId); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeFixedString(os, subjectPos.getWorld(), 24); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeFloat64(os, subjectPos.getX()); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeFloat64(os, subjectPos.getY()); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeFloat64(os, subjectPos.getZ()); - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeFixedString(os, subjectUUID.toString(), 36); // lowercase - BinaryStreamUtils.writeNonNull(os); BinaryStreamUtils.writeInt64(os, timeMillis); } } diff --git a/src/main/java/com/keuin/blame/lookup/AbstractLookupFilter.java b/src/main/java/com/keuin/blame/lookup/AbstractLookupFilter.java deleted file mode 100644 index f2e45dc..0000000 --- a/src/main/java/com/keuin/blame/lookup/AbstractLookupFilter.java +++ /dev/null @@ -1,7 +0,0 @@ -package com.keuin.blame.lookup; - -import org.bson.conversions.Bson; - -public interface AbstractLookupFilter { - abstract Bson filter(); -}
\ No newline at end of file diff --git a/src/main/java/com/keuin/blame/test/TestDatabase.java b/src/main/java/com/keuin/blame/test/TestDatabase.java index 84bf4bc..6d40eb5 100644 --- a/src/main/java/com/keuin/blame/test/TestDatabase.java +++ b/src/main/java/com/keuin/blame/test/TestDatabase.java @@ -6,9 +6,6 @@ import com.keuin.blame.data.WorldPos; import com.keuin.blame.data.entry.LogEntry; import com.keuin.blame.data.enums.ActionType; import com.keuin.blame.data.enums.ObjectType; -import com.keuin.blame.lookup.DummyFilter; -import com.keuin.blame.lookup.LookupCallback; -import com.keuin.blame.lookup.LookupManager; import org.junit.Before; import org.junit.Test; @@ -70,15 +67,15 @@ public class TestDatabase { ); SubmitWorker.INSTANCE.submit(entry); Thread.sleep(2000); - LookupManager.INSTANCE.lookup(new DummyFilter(), new LookupCallback() { - @Override - public void onLookupFinishes(Iterable<LogEntry> logEntries) { - for (LogEntry e : logEntries) { - System.out.println(e); - success[0] |= Objects.equals(e, entry); - } - } - }, 100); +// LookupManager.INSTANCE.lookup(new DummyFilter(), new LookupCallback() { +// @Override +// public void onLookupFinishes(Iterable<LogEntry> logEntries) { +// for (LogEntry e : logEntries) { +// System.out.println(e); +// success[0] |= Objects.equals(e, entry); +// } +// } +// }, 100); Thread.sleep(2000); assertTrue(success[0]); } catch (Exception e) { |