class KVOperation {

    /**
     * Put operation
     */
    public static final byte PUT = 0x01;
    /**
     * Get operation
     */
    public static final byte GET = 0x02;
    public static final byte DEL = 0x03;
    public static final byte REMOVE_BY_PREFIX = 0x04;
    public static final byte REMOVE = 0x05;
    public static final byte PUT_WITH_TTL = 0x06;
    public static final byte CLEAR = 0x07;
    public static final byte PUT_WITH_TTL_UNIT = 0x08;
    public static final byte REMOVE_WITH_TTL = 0x09;
    /**
     * Snapshot operation
     */
    public static final byte SAVE_SNAPSHOT = 0x10;
    public static final byte LOAD_SNAPSHOT = 0x11;

    private byte[] key;
    private byte[] value;
    private Object attach; // 原始对象，用于本机处理，减少一次反序列化操作
    private Object arg;
    private byte op;

    public KVOperation() {

    }

    public KVOperation(byte[] key, byte[] value, Object attach, byte op) {
        this.key = key;
        this.value = value;
        this.attach = attach;
        this.op = op;
    }

    public KVOperation(byte[] key, byte[] value, Object attach, byte op, Object arg) {
        this.key = key;
        this.value = value;
        this.attach = attach;
        this.op = op;
        this.arg = arg;
    }

    public static KVOperation fromByteArray(byte[] value) throws IOException {

        try (ByteArrayInputStream bis = new ByteArrayInputStream(value, 1, value.length - 1)) {
            Hessian2Input input = new Hessian2Input(bis);
            KVOperation op = new KVOperation();
            op.op = value[0];
            op.key = input.readBytes();
            op.value = input.readBytes();
            op.arg = input.readObject();
            input.close();
            return op;
        }
    }

    public static KVOperation createPut(final byte[] key, final byte[] value) {
        Requires.requireNonNull(key, "key");
        Requires.requireNonNull(value, "value");
        return new KVOperation(key, value, null, PUT);
    }

    public static KVOperation createGet(final byte[] key) {
        Requires.requireNonNull(key, "key");
        return new KVOperation(key, BytesUtil.EMPTY_BYTES, null, GET);
    }

    public static KVOperation createPutWithTTL(byte[] key, byte[] value, long ttl) {<FILL_FUNCTION_BODY>}

    public static KVOperation createPutWithTTL(byte[] key, byte[] value, long ttl,
                                               TimeUnit timeUnit) {
        Requires.requireNonNull(key, "key");
        Requires.requireNonNull(value, "value");
        return new KVOperation(key, value, value, PUT_WITH_TTL_UNIT,
                               new Object[]{ttl, timeUnit});
    }

    public static KVOperation createRemoveWithTTL(byte[] key) {
        Requires.requireNonNull(key, "key");
        return new KVOperation(key, key, null, REMOVE_WITH_TTL);
    }

    public static KVOperation createRemoveByPrefix(byte[] key) {
        Requires.requireNonNull(key, "key");
        return new KVOperation(key, key, null, REMOVE_BY_PREFIX);
    }

    public static KVOperation createRemove(byte[] key) {
        Requires.requireNonNull(key, "key");
        return new KVOperation(key, key, null, REMOVE);
    }

    public static KVOperation createClear() {
        return new KVOperation(null, null, null, CLEAR);
    }

    public static KVOperation createSaveSnapshot(String snapshotPath) {
        Requires.requireNonNull(snapshotPath, "snapshotPath");
        return new KVOperation(null, null, snapshotPath, SAVE_SNAPSHOT);
    }

    public static KVOperation createLoadSnapshot(String snapshotPath) {
        Requires.requireNonNull(snapshotPath, "snapshotPath");
        return new KVOperation(null, null, snapshotPath, LOAD_SNAPSHOT);
    }

    public byte[] toByteArray() throws IOException {
        try (ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
            bos.write(op);
            Hessian2Output output = new Hessian2Output(bos);
            output.writeObject(key);
            output.writeObject(value);
            output.writeObject(arg);
            output.flush();
            return bos.toByteArray();
        }
    }
}

        Requires.requireNonNull(key, "key");
        Requires.requireNonNull(value, "value");
        return new KVOperation(key, value, value, PUT_WITH_TTL,
                               ttl);
    
<no_super_class>