rm not used

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/impl/BinlogColumnValue.java → dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/BinlogColumnValue.java

@@ -1,8 +1,7 @@
-package org.dbsyncer.storage.binlog.impl;
+package org.dbsyncer.storage.binlog;
 
 import com.google.protobuf.ByteString;
 import org.dbsyncer.common.column.AbstractColumnValue;
-import org.dbsyncer.common.util.DateFormatUtil;
 
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/BinlogContext.java

@@ -1,361 +0,0 @@
-package org.dbsyncer.storage.binlog;
-
-import org.apache.commons.io.FileUtils;
-import org.dbsyncer.common.scheduled.ScheduledTaskJob;
-import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.common.util.JsonUtil;
-import org.dbsyncer.common.util.NumberUtil;
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.storage.binlog.impl.BinlogPipeline;
-import org.dbsyncer.storage.binlog.impl.BinlogReader;
-import org.dbsyncer.storage.binlog.impl.BinlogWriter;
-import org.dbsyncer.storage.binlog.proto.BinlogMessage;
-import org.dbsyncer.storage.model.BinlogConfig;
-import org.dbsyncer.storage.model.BinlogIndex;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.util.Assert;
-
-import java.io.Closeable;
-import java.io.File;
-import java.io.IOException;
-import java.nio.charset.Charset;
-import java.nio.file.Files;
-import java.nio.file.attribute.BasicFileAttributes;
-import java.time.LocalDateTime;
-import java.time.ZoneId;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-/**
- * <p>组件介绍</p>
- * <ol>
- *     <li>BinlogPipeline（提供文件流读写）</li>
- *     <li>定时器（维护索引文件状态，回收文件流）</li>
- *     <li>索引</li>
- * </ol>
- * <p>定时器</p>
- * <ol>
- *     <li>生成新索引（超过限制大小｜过期）</li>
- *     <li>关闭索引流（有锁 & 读写状态关闭 & 30s未用）</li>
- *     <li>删除旧索引（无锁 & 过期）</li>
- * </ol>
- *
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/29 1:28
- */
-public class BinlogContext implements ScheduledTaskJob, Closeable {
-
-    private final Logger logger = LoggerFactory.getLogger(getClass());
-
-    private static final long BINLOG_MAX_SIZE = 256 * 1024 * 1024;
-
-    private static final int BINLOG_EXPIRE_DAYS = 7;
-
-    private static final int BINLOG_ACTUATOR_CLOSE_DELAYED_SECONDS = 30;
-
-    private static final String LINE_SEPARATOR = System.lineSeparator();
-
-    private static final Charset DEFAULT_CHARSET = Charset.defaultCharset();
-
-    private static final String BINLOG = "binlog";
-
-    private static final String BINLOG_INDEX = BINLOG + ".index";
-
-    private static final String BINLOG_CONFIG = BINLOG + ".config";
-
-    private final List<BinlogIndex> indexList = new LinkedList<>();
-
-    private final String path;
-
-    private final File configFile;
-
-    private final File indexFile;
-
-    private final BinlogPipeline pipeline;
-
-    private final Lock readerLock = new ReentrantLock(true);
-
-    private final Lock lock = new ReentrantLock(true);
-
-    private volatile boolean running;
-
-    private BinlogConfig config;
-
-    public BinlogContext(String taskName) throws IOException {
-        path = new StringBuilder(System.getProperty("user.dir")).append(File.separatorChar)
-                .append("data").append(File.separatorChar)
-                .append("binlog").append(File.separatorChar)
-                .append(taskName).append(File.separatorChar)
-                .toString();
-        File dir = new File(path);
-        if (!dir.exists()) {
-            FileUtils.forceMkdir(dir);
-        }
-
-        // binlog.index
-        indexFile = new File(path + BINLOG_INDEX);
-        // binlog.config
-        configFile = new File(path + BINLOG_CONFIG);
-        if (!configFile.exists()) {
-            // binlog.000001
-            config = initBinlogConfigAndIndex(createNewBinlogName(0));
-        }
-
-        // read index
-        Assert.isTrue(indexFile.exists(), String.format("The index file '%s' is not exist.", indexFile.getName()));
-        readIndexFromDisk();
-
-        // delete index file
-        deleteExpiredIndexFile();
-
-        // {"binlog":"binlog.000001","pos":0}
-        if (null == config) {
-            config = JsonUtil.jsonToObj(FileUtils.readFileToString(configFile, DEFAULT_CHARSET), BinlogConfig.class);
-        }
-
-        // no index
-        if (CollectionUtils.isEmpty(indexList)) {
-            // binlog.000002
-            config = initBinlogConfigAndIndex(createNewBinlogName(config.getFileName()));
-            readIndexFromDisk();
-        }
-
-        // 配置文件已失效，取最早的索引文件
-        BinlogIndex startBinlogIndex = getBinlogIndexByFileName(config.getFileName());
-        if (null == startBinlogIndex) {
-            logger.warn("The binlog file '{}' is expired.", config.getFileName());
-            startBinlogIndex = indexList.get(0);
-            config = new BinlogConfig().setFileName(startBinlogIndex.getFileName());
-            write(configFile, JsonUtil.objToJson(config), false);
-        }
-
-        final BinlogWriter binlogWriter = new BinlogWriter(path, indexList.get(indexList.size() - 1));
-        final BinlogReader binlogReader = new BinlogReader(path, startBinlogIndex, config.getPosition());
-        pipeline = new BinlogPipeline(binlogWriter, binlogReader);
-        logger.info("BinlogContext initialized with config:{}", JsonUtil.objToJson(config));
-    }
-
-    @Override
-    public void run() {
-        if (running) {
-            return;
-        }
-
-        final Lock binlogLock = lock;
-        boolean locked = false;
-        try {
-            locked = binlogLock.tryLock();
-            if (locked) {
-                running = true;
-                createNewBinlogIndex();
-                closeFreeBinlogIndex();
-                deleteOldBinlogIndex();
-            }
-        } catch (Exception e) {
-            logger.error(e.getMessage());
-        } finally {
-            if (locked) {
-                running = false;
-                binlogLock.unlock();
-            }
-        }
-    }
-
-    @Override
-    public void close() {
-        pipeline.close();
-    }
-
-    public void flush() throws IOException {
-        config.setFileName(pipeline.getReaderFileName());
-        config.setPosition(pipeline.getReaderOffset());
-        write(configFile, JsonUtil.objToJson(config), false);
-    }
-
-    public byte[] readLine() throws IOException {
-        byte[] line = pipeline.readLine();
-        if(null == line){
-            switchNextBinlogIndex();
-        }
-        return line;
-    }
-
-    public void write(BinlogMessage message) throws IOException {
-        pipeline.write(message);
-    }
-
-    public BinlogIndex getBinlogIndexByFileName(String fileName) {
-        BinlogIndex index = null;
-        for (BinlogIndex binlogIndex : indexList) {
-            if (StringUtil.equals(binlogIndex.getFileName(), fileName)) {
-                index = binlogIndex;
-                break;
-            }
-        }
-        return index;
-    }
-
-    private void switchNextBinlogIndex() {
-        // 有新索引文件
-        if(!isCreatedNewBinlogIndex()){
-            return;
-        }
-        boolean locked = false;
-        try {
-            locked = readerLock.tryLock();
-            if (locked) {
-                // 有新索引文件
-                if(isCreatedNewBinlogIndex()){
-                    String newBinlogName = createNewBinlogName(pipeline.getReaderFileName());
-                    BinlogIndex startBinlogIndex = getBinlogIndexByFileName(newBinlogName);
-                    final BinlogReader binlogReader = pipeline.getBinlogReader();
-                    config = new BinlogConfig().setFileName(startBinlogIndex.getFileName());
-                    write(configFile, JsonUtil.objToJson(config), false);
-                    pipeline.setBinlogReader(new BinlogReader(path, startBinlogIndex, config.getPosition()));
-                    binlogReader.stop();
-                    logger.info("Switch to new index file '{}' for binlog reader.", newBinlogName);
-                }
-            }
-        } catch (Exception e) {
-            logger.error(e.getMessage());
-        } finally {
-            if (locked) {
-                readerLock.unlock();
-            }
-        }
-    }
-
-    private boolean isCreatedNewBinlogIndex() {
-        return !StringUtil.equals(pipeline.getReaderFileName(), pipeline.getWriterFileName());
-    }
-
-    private void createNewBinlogIndex() throws IOException {
-        final String writerFileName = pipeline.getWriterFileName();
-        File file = new File(path + writerFileName);
-        // 超过限制大小｜过期
-        if (file.length() > BINLOG_MAX_SIZE || isExpiredFile(file)) {
-            final BinlogWriter binlogWriter = pipeline.getBinlogWriter();
-            String newBinlogName = createNewBinlogName(writerFileName);
-            logger.info("The size of index file '{}' has reached {}MB, exceeding the limit of {}MB, switching to a new index file '{}' for index writer.", writerFileName, getMB(file.length()), getMB(BINLOG_MAX_SIZE), newBinlogName);
-            write(indexFile, newBinlogName + LINE_SEPARATOR, true);
-            File binlogIndexFile = new File(path + newBinlogName);
-            write(binlogIndexFile, "", false);
-            BinlogIndex newBinlogIndex = new BinlogIndex(newBinlogName, getFileCreateDateTime(binlogIndexFile));
-            indexList.add(newBinlogIndex);
-            pipeline.setBinlogWriter(new BinlogWriter(path, newBinlogIndex));
-            binlogWriter.stop();
-        }
-    }
-
-    private void closeFreeBinlogIndex() throws IOException {
-        Iterator<BinlogIndex> iterator = indexList.iterator();
-        while (iterator.hasNext()) {
-            BinlogIndex next = iterator.next();
-            // 有锁 & 读写状态关闭 & 30s未用
-            if (!next.isFreeLock() && !next.isRunning() && next.getUpdateTime().isBefore(LocalDateTime.now().minusSeconds(BINLOG_ACTUATOR_CLOSE_DELAYED_SECONDS))) {
-                next.removeAllLock();
-                logger.info("Close free index file '{}', the last update time is {}", next.getFileName(), next.getUpdateTime());
-            }
-        }
-    }
-
-    private void deleteOldBinlogIndex() throws IOException {
-        Iterator<BinlogIndex> iterator = indexList.iterator();
-        while (iterator.hasNext()) {
-            BinlogIndex next = iterator.next();
-            // 无锁 & 过期
-            if (next.isFreeLock()) {
-                File file = new File(path + next.getFileName());
-                if(isExpiredFile(file)){
-                    FileUtils.forceDelete(file);
-                    iterator.remove();
-                }
-            }
-        }
-    }
-
-    private long getMB(long size) {
-        return size / 1024 / 1024;
-    }
-
-    private void readIndexFromDisk() throws IOException {
-        indexList.clear();
-        List<String> indexNames = FileUtils.readLines(indexFile, DEFAULT_CHARSET);
-        if (!CollectionUtils.isEmpty(indexNames)) {
-            for (String indexName : indexNames) {
-                File file = new File(path + indexName);
-                if(file.exists()){
-                    indexList.add(new BinlogIndex(indexName, getFileCreateDateTime(file)));
-                }
-            }
-        }
-    }
-
-    private BinlogConfig initBinlogConfigAndIndex(String binlogName) throws IOException {
-        BinlogConfig config = new BinlogConfig().setFileName(binlogName);
-        write(configFile, JsonUtil.objToJson(config), false);
-        write(indexFile, binlogName + LINE_SEPARATOR, false);
-        write(new File(path + binlogName), "", false);
-        return config;
-    }
-
-    private void deleteExpiredIndexFile() throws IOException {
-        if (CollectionUtils.isEmpty(indexList)) {
-            return;
-        }
-        boolean delete = false;
-        Iterator<BinlogIndex> iterator = indexList.iterator();
-        while (iterator.hasNext()) {
-            BinlogIndex next = iterator.next();
-            if (null == next) {
-                continue;
-            }
-            File file = new File(path + next.getFileName());
-            if (!file.exists()) {
-                logger.info("Delete invalid binlog file '{}'.", next.getFileName());
-                iterator.remove();
-                delete = true;
-                continue;
-            }
-            if (isExpiredFile(file)) {
-                FileUtils.forceDelete(file);
-                iterator.remove();
-                delete = true;
-                logger.info("Delete expired binlog file '{}'.", next.getFileName());
-            }
-        }
-
-        if (delete) {
-            StringBuilder indexBuilder = new StringBuilder();
-            indexList.forEach(i -> indexBuilder.append(i.getFileName()).append(LINE_SEPARATOR));
-            write(indexFile, indexBuilder.toString(), false);
-        }
-    }
-
-    private boolean isExpiredFile(File file) throws IOException {
-        final LocalDateTime createTime = getFileCreateDateTime(file);
-        return createTime.isBefore(LocalDateTime.now().minusDays(BINLOG_EXPIRE_DAYS));
-    }
-
-    private LocalDateTime getFileCreateDateTime(File file) throws IOException {
-        BasicFileAttributes attr = Files.readAttributes(file.toPath(), BasicFileAttributes.class);
-        return attr.creationTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
-    }
-
-    private String createNewBinlogName(int index) {
-        return String.format("%s.%06d", BINLOG, index % 999999 + 1);
-    }
-
-    private String createNewBinlogName(String binlogName) {
-        return createNewBinlogName(NumberUtil.toInt(StringUtil.substring(binlogName, BINLOG.length() + 1)));
-    }
-
-    private void write(File file, String line, boolean append) throws IOException {
-        FileUtils.write(file, line, DEFAULT_CHARSET, append);
-    }
-}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/impl/BinlogPipeline.java

@@ -1,63 +0,0 @@
-package org.dbsyncer.storage.binlog.impl;
-
-import org.dbsyncer.storage.binlog.proto.BinlogMessage;
-
-import java.io.Closeable;
-import java.io.IOException;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/19 23:36
- */
-public class BinlogPipeline implements Closeable {
-    private BinlogWriter binlogWriter;
-    private BinlogReader binlogReader;
-
-    public BinlogPipeline(BinlogWriter binlogWriter, BinlogReader binlogReader) {
-        this.binlogWriter = binlogWriter;
-        this.binlogReader = binlogReader;
-    }
-
-    public void write(BinlogMessage message) throws IOException {
-        binlogWriter.write(message);
-    }
-
-    public byte[] readLine() throws IOException {
-        return binlogReader.readLine();
-    }
-
-    public long getReaderOffset() {
-        return binlogReader.getOffset();
-    }
-
-    public String getReaderFileName() {
-        return binlogReader.getFileName();
-    }
-
-    public String getWriterFileName() {
-        return binlogWriter.getFileName();
-    }
-
-    @Override
-    public void close() {
-        binlogWriter.close();
-        binlogReader.close();
-    }
-
-    public BinlogWriter getBinlogWriter() {
-        return binlogWriter;
-    }
-
-    public void setBinlogWriter(BinlogWriter binlogWriter) {
-        this.binlogWriter = binlogWriter;
-    }
-
-    public BinlogReader getBinlogReader() {
-        return binlogReader;
-    }
-
-    public void setBinlogReader(BinlogReader binlogReader) {
-        this.binlogReader = binlogReader;
-    }
-}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/impl/BinlogReader.java

@@ -1,53 +0,0 @@
-package org.dbsyncer.storage.binlog.impl;
-
-import com.google.protobuf.CodedInputStream;
-import org.apache.commons.io.IOUtils;
-import org.dbsyncer.common.file.BufferedRandomAccessFile;
-import org.dbsyncer.storage.binlog.AbstractBinlogActuator;
-import org.dbsyncer.storage.model.BinlogIndex;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.RandomAccessFile;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/26 23:25
- */
-public class BinlogReader extends AbstractBinlogActuator {
-    private final RandomAccessFile raf;
-    private final byte[] h = new byte[4];
-    private byte[] b;
-    private long offset;
-    private CodedInputStream cis;
-
-    public BinlogReader(String path, BinlogIndex binlogIndex, long position) throws IOException {
-        initBinlogIndex(binlogIndex);
-        this.raf = new BufferedRandomAccessFile(new File(path + binlogIndex.getFileName()), "r");
-        raf.seek(position);
-    }
-
-    public byte[] readLine() throws IOException {
-        this.offset = raf.getFilePointer();
-        if (offset >= raf.length()) {
-            return null;
-        }
-        raf.read(h);
-        cis = CodedInputStream.newInstance(h);
-        b = new byte[cis.readFixed32()];
-        raf.read(b);
-        raf.seek(this.offset + (h.length + b.length));
-        refreshBinlogIndexUpdateTime();
-        return b;
-    }
-
-    public long getOffset() {
-        return offset;
-    }
-
-    @Override
-    public void close() {
-        IOUtils.closeQuietly(raf);
-    }
-}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/impl/BinlogWriter.java

@@ -1,45 +0,0 @@
-package org.dbsyncer.storage.binlog.impl;
-
-import com.google.protobuf.CodedOutputStream;
-import org.apache.commons.io.IOUtils;
-import org.dbsyncer.storage.binlog.AbstractBinlogActuator;
-import org.dbsyncer.storage.binlog.proto.BinlogMessage;
-import org.dbsyncer.storage.model.BinlogIndex;
-
-import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/26 23:28
- */
-public class BinlogWriter extends AbstractBinlogActuator {
-
-    private final OutputStream out;
-
-    public BinlogWriter(String path, BinlogIndex binlogIndex) throws FileNotFoundException {
-        initBinlogIndex(binlogIndex);
-        this.out = new FileOutputStream(path + binlogIndex.getFileName(), true);
-    }
-
-    public void write(BinlogMessage message) throws IOException {
-        if(null != message){
-            // 选择固定长度int32作为tag标志位，4bytes, 最多可容纳2^31-1字节（2047MB左右, 建议上限64~128M内最佳）,
-            final int serialized = message.getSerializedSize();
-            final int bufferSize = CodedOutputStream.computeFixed32SizeNoTag(serialized) + serialized;
-            final CodedOutputStream codedOutput = CodedOutputStream.newInstance(out, bufferSize);
-            codedOutput.writeFixed32NoTag(serialized);
-            message.writeTo(codedOutput);
-            codedOutput.flush();
-            refreshBinlogIndexUpdateTime();
-        }
-    }
-
-    @Override
-    public void close() {
-        IOUtils.closeQuietly(out);
-    }
-}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/util/BinlogMessageUtil.java

@@ -5,7 +5,7 @@ import oracle.sql.BLOB;
 import oracle.sql.CLOB;
 import oracle.sql.TIMESTAMP;
 import org.apache.commons.io.IOUtils;
-import org.dbsyncer.storage.binlog.impl.BinlogColumnValue;
+import org.dbsyncer.storage.binlog.BinlogColumnValue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

dbsyncer-storage/src/main/test/BinlogMessageTest.java

@@ -1,227 +0,0 @@
-import com.google.protobuf.ByteString;
-import com.google.protobuf.InvalidProtocolBufferException;
-import org.apache.commons.io.IOUtils;
-import org.dbsyncer.storage.binlog.AbstractBinlogRecorder;
-import org.dbsyncer.storage.binlog.BinlogContext;
-import org.dbsyncer.storage.binlog.impl.BinlogColumnValue;
-import org.dbsyncer.storage.binlog.proto.BinlogMap;
-import org.dbsyncer.storage.binlog.proto.BinlogMessage;
-import org.dbsyncer.storage.binlog.proto.EventEnum;
-import org.dbsyncer.storage.util.BinlogMessageUtil;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.nio.charset.Charset;
-import java.sql.Date;
-import java.sql.Time;
-import java.sql.Timestamp;
-import java.time.LocalDateTime;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Queue;
-
-import static org.dbsyncer.storage.util.BinlogMessageUtil.serializeValue;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/18 23:46
- */
-public class BinlogMessageTest extends AbstractBinlogRecorder {
-
-    private final Logger logger = LoggerFactory.getLogger(getClass());
-
-    private BinlogContext context;
-
-    private BinlogColumnValue value = new BinlogColumnValue();
-
-    @Before
-    public void init() throws IOException {
-        context = new BinlogContext("WriterBinlog");
-    }
-
-    @After
-    public void close() {
-        IOUtils.closeQuietly(context);
-    }
-
-    @Test
-    public void testBinlogMessage() throws IOException {
-        for (int i = 0; i < 10000; i++) {
-            write("123456", i+"");
-        }
-        //write("000111", "xyz");
-        //write("888999", "jkl");
-
-        byte[] line;
-        int count = 0;
-        while (null != (line = context.readLine())) {
-            //logger.info("size:{}, {}", line.length, line);
-            try {
-                BinlogMessage message = BinlogMessage.parseFrom(line);
-                if(null != message){
-                    count ++;
-                    message.getData();
-                }
-            } catch (InvalidProtocolBufferException e) {
-                logger.info("{} : {}", line.length, line);
-            }
-        }
-        logger.info("总条数：{}", count);
-        context.flush();
-    }
-
-    private void write(String tableGroupId, String key) throws IOException {
-        Map<String, Object> data = new HashMap<>();
-        data.put("id", 1L);
-        data.put("name", key + "中文");
-        data.put("age", 88);
-        data.put("bd", new BigDecimal(88));
-        data.put("sex", 1);
-        data.put("f", 88.88f);
-        data.put("d", 999.99d);
-        data.put("b", true);
-        short ss = 32767;
-        data.put("ss", ss);
-        data.put("bytes", "中文666".getBytes(Charset.defaultCharset()));
-        data.put("create_date", new Date(Timestamp.valueOf(LocalDateTime.now()).getTime()));
-        data.put("update_time", Timestamp.valueOf(LocalDateTime.now()).getTime());
-
-        BinlogMap.Builder builder = BinlogMap.newBuilder();
-        data.forEach((k, v) -> {
-            if (null != v) {
-                ByteString bytes = serializeValue(v);
-                if (null != bytes) {
-                    builder.putRow(k, bytes);
-                }
-            }
-        });
-
-        BinlogMessage build = BinlogMessage.newBuilder()
-                .setTableGroupId(tableGroupId)
-                .setEvent(EventEnum.UPDATE)
-                .setData(builder.build())
-                .build();
-        //byte[] bytes = build.toByteArray();
-        //logger.info("序列化长度：{}", bytes.length);
-        //logger.info("{}", bytes);
-        context.write(build);
-    }
-
-    @Test
-    public void testMessageNumber() {
-        // short
-        short s = 32767;
-        logger.info("short1:{}", s);
-        ByteString shortBytes = serializeValue(s);
-        logger.info("bytes:{}", shortBytes.toByteArray());
-        value.setValue(shortBytes);
-        short s2 = value.asShort();
-        logger.info("short2:{}", s2);
-
-        // int
-        int i = 1999999999;
-        logger.info("int1:{}", i);
-        ByteString intBytes = serializeValue(i);
-        logger.info("bytes:{}", intBytes.toByteArray());
-        value.setValue(intBytes);
-        int i2 = value.asInteger();
-        logger.info("int2:{}", i2);
-
-        // long
-        long l = 8999999999999999999L;
-        logger.info("long1:{}", l);
-        ByteString longBytes = serializeValue(l);
-        logger.info("bytes:{}", longBytes.toByteArray());
-        value.setValue(longBytes);
-        long l2 = value.asLong();
-        logger.info("long2:{}", l2);
-
-        // float
-        float f = 99999999999999999999999999999999999.99999999999999999999999999999999999f;
-        logger.info("float1:{}", f);
-        ByteString floatBytes = serializeValue(f);
-        logger.info("bytes:{}", floatBytes.toByteArray());
-        value.setValue(floatBytes);
-        float f2 = value.asFloat();
-        logger.info("float2:{}", f2);
-
-        // double
-        double d = 999999.9999999999999999999999999d;
-        logger.info("double1:{}", d);
-        ByteString doubleBytes = serializeValue(d);
-        logger.info("bytes:{}", doubleBytes.toByteArray());
-        value.setValue(doubleBytes);
-        double d2 = value.asDouble();
-        logger.info("double2:{}", d2);
-
-        // double
-        BigDecimal b = new BigDecimal(8888888.888888888888888f);
-        logger.info("bigDecimal1:{}", b);
-        ByteString bigDecimalBytes = serializeValue(b);
-        logger.info("bytes:{}", bigDecimalBytes.toByteArray());
-        value.setValue(bigDecimalBytes);
-        BigDecimal b2 = value.asBigDecimal();
-        logger.info("bigDecimal2:{}", b2);
-
-        // boolean
-        boolean bool = true;
-        logger.info("bool1:{}", bool);
-        ByteString boolBytes = serializeValue(bool);
-        logger.info("bytes:{}", boolBytes.toByteArray());
-        value.setValue(boolBytes);
-        Boolean bool2 = value.asBoolean();
-        logger.info("bool2:{}", bool2);
-    }
-
-    @Test
-    public void testMessageDate() {
-        // timestamp
-        Timestamp timestamp = Timestamp.valueOf(LocalDateTime.now());
-        logger.info("timestamp1:{}, l:{}", timestamp, timestamp.getTime());
-        ByteString timestampBytes = serializeValue(timestamp);
-        logger.info("bytes:{}", timestampBytes.toByteArray());
-        value.setValue(timestampBytes);
-        Timestamp timestamp2 = value.asTimestamp();
-        logger.info("timestamp2:{}, l:{}", timestamp2, timestamp2.getTime());
-
-        // date
-        Date date = new Date(timestamp.getTime());
-        logger.info("date1:{}, l:{}", date, date.getTime());
-        ByteString dateBytes = serializeValue(date);
-        logger.info("bytes:{}", dateBytes.toByteArray());
-        value.setValue(dateBytes);
-        Date date2 = value.asDate();
-        logger.info("date2:{}, l:{}", date2, date2.getTime());
-
-        // time
-        Time time = new Time(timestamp.getTime());
-        logger.info("time1:{}, l:{}", time, time.getTime());
-        ByteString timeBytes = serializeValue(time);
-        logger.info("bytes:{}", timeBytes.toByteArray());
-        value.setValue(timeBytes);
-        Time time2 = value.asTime();
-        logger.info("time2:{}, l:{}", time2, time2.getTime());
-    }
-
-    @Override
-    public Queue getQueue() {
-        return null;
-    }
-
-    @Override
-    public int getQueueCapacity() {
-        return 0;
-    }
-
-    @Override
-    protected Object deserialize(String messageId, BinlogMessage message) {
-        return null;
-    }
-
-}

dbsyncer-storage/src/main/test/ShardBinlogTest.java

@@ -8,7 +8,6 @@ import org.apache.lucene.search.*;
 import org.apache.lucene.util.BytesRef;
 import org.dbsyncer.common.model.Paging;
 import org.dbsyncer.common.util.DateFormatUtil;
-import org.dbsyncer.storage.binlog.AbstractBinlogRecorder;
 import org.dbsyncer.storage.binlog.proto.BinlogMap;
 import org.dbsyncer.storage.binlog.proto.BinlogMessage;
 import org.dbsyncer.storage.binlog.proto.EventEnum;
@@ -31,7 +30,10 @@ import java.sql.Date;
 import java.sql.Timestamp;
 import java.time.Instant;
 import java.time.LocalDateTime;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -39,7 +41,7 @@ import java.util.concurrent.TimeUnit;
  * @version 1.0.0
  * @date 2022/6/18 23:46
  */
-public class ShardBinlogTest extends AbstractBinlogRecorder {
+public class ShardBinlogTest {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -97,7 +99,7 @@ public class ShardBinlogTest extends AbstractBinlogRecorder {
         List<Document> list = new ArrayList<>();
         long now = Instant.now().toEpochMilli();
         int size = i + 5;
-        while (i < size){
+        while (i < size) {
             BinlogMessage message = genMessage("123456", i + "");
             BytesRef bytes = new BytesRef(message.toByteArray());
             list.add(ParamsUtil.convertBinlog2Doc(String.valueOf(i), BinlogConstant.READY, bytes, now));
@@ -182,21 +184,6 @@ public class ShardBinlogTest extends AbstractBinlogRecorder {
         return build;
     }
 
-    @Override
-    public Queue getQueue() {
-        return null;
-    }
-
-    @Override
-    public int getQueueCapacity() {
-        return 0;
-    }
-
-    @Override
-    protected Object deserialize(String messageId, BinlogMessage message) {
-        return null;
-    }
-
     private void check() throws IOException {
         final IndexSearcher searcher = shard.getSearcher();
         IndexReader reader = searcher.getIndexReader();