# YG4，client get command before propagation set command， so get ""

**URL:** <https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610>\
**Category:** Challenges\
**Tags:** challenge:redis\
**Created:** [March 13, 2025, 3:55am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610 "2025-03-13T03:55:38Z")\
**Posts on this page:** 20\
**Page:** 1

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 13, 2025, 3:55am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/1 "2025-03-13T03:55:38Z")

</div>

I’m stuck on Stage #(YG4).

Hello, guys, my question is about YG4. I will post my Java code and logs.  
Here is my issue: I have started a separate thread specifically for handling handshake and propagation, and I have also started threads to handle client requests. This means there is a multithreading issue here because I cannot control the timing of propagation and client requests As a result, I cannot pass the YG4 test because when the client issues a get command, my propagation thread has not yet completed the set command. This is evident from the logs, where there is a time gap between the set and get commands. If I make the client thread handling the get command sleep for a while, it works, but I believe this is inappropriate.  
how can I ensure that the propagation set command is executed before the get command?

help me please， thank you guys .

Here are my logs:

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/d/d295d5f2a265c5b9108d1d68a9eb93d27f6b28ca.jpeg)  
 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/d/d1a7419f11b58ed43aed33d365b0223a35b8c4f4.png)  
 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/2/2167483980e6f3f261be38b4891d3171b35bb4f5.png)

And here’s my code:

```Java
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

class KeyExpiry {
    public long expTime;
    public boolean isActive;
    public long startTime;

    public KeyExpiry(long expTime, boolean isActive, long startTime) {
        this.expTime = expTime;
        this.isActive = isActive;
        this.startTime = startTime;
    }
}

public class Main {
    private static HashSet<Socket> replicaSockets = new HashSet<>();
    private static Map<String, String> map;
    private static Map<String, KeyExpiry> expMap;
    private static String currentKey = "";
    private static int port = 6379;
    private static boolean isMaster = true;

    public static void main(String[] args) {

        map = new ConcurrentHashMap<>();
        expMap = new ConcurrentHashMap<>();
        // You can use print statements as follows for debugging, they'll be visible when running tests.
        System.out.println("Logs from your program will appear here!");

        // Uncomment this block to pass the first stage
        ServerSocket serverSocket = null;

        handleArgs(List.of(args));
        try {
            serverSocket = new ServerSocket(port);

            // Since the tester restarts your program quite often, setting SO_REUSEADDR
            // ensures that we don't run into 'Address already in use' errors
            serverSocket.setReuseAddress(true);
            // Wait for connection from client.
            while (true) {
                Socket clientSocket = serverSocket.accept();

                new Thread(() -> {
                    try {
                        processClients(clientSocket);
                    } catch (Exception e) {
                        System.out.println("Exception: " + e.getMessage());
                    }
                }).start();
            }

        } catch (IOException e) {
            System.out.println("IOException: " + e.getMessage());
        }
    }

    private static void handleArgs(List<String> args) {
        if (args.contains("--port")) {
            port = Integer.parseInt(args.get(args.indexOf("--port") + 1));
        }
        if (args.contains("--replicaof")) {
            String[] replicaArgs = args.get(args.indexOf("--replicaof") + 1).split(" ");
            String masterHostname = replicaArgs[0];
            int masterPortNumber = Integer.parseInt(replicaArgs[1]);
            isMaster = false;
            replicaHandshake(masterHostname, masterPortNumber);
        }
    }

    private static void replicaHandshake(String masterHostname, int masterPortNumber) {
        try {
            new Thread(() -> {
                try {
                    processMasterPropagation(masterHostname,masterPortNumber);
                } catch (Exception e) {
                    System.out.println("Exception: " + e.getMessage());
                }
            }).start();

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static void sendToReplica(Vector<String> command) throws IOException {
        String toSend = "";
        String arr[] = new String[command.size()];
        for (int i = 0; i < command.size(); i++) {
            arr[i] = command.get(i);
        }
        toSend = encodeRESPArr(arr);
        for (Socket s : replicaSockets) {
            OutputStream os = s.getOutputStream();
            System.out.println("Writing to a replica");
            os.write(toSend.getBytes());
        }
    }

    public static String encodeRESPArr(String[] arr) {
        int n = arr.length;
        String send = "*" + n + "\r\n";
        for (String s : arr) {
            send += encodeRESP(s);
        }
        return send;
    }

    public static String encodeRESP(String s) {
        int size = s.length();
        String ret = "$" + size + "\r\n" + s + "\r\n";
        return ret;
    }

    private static void processMasterPropagation(String masterHostname , int masterPortNumber) {

        try {
            String request_ping = "*1\r\n$4\r\nping\r\n";
            String request_replconf_1 = "*3\r\n$8\r\nREPLCONF\r\n$14\r\nlistening-port\r\n$4\r\n6380\r\n";
            String request_replconf_2 = "*3\r\n$8\r\nREPLCONF\r\n$4\r\ncapa\r\n$6\r\npsync2\r\n";
            String request_PSYNC = "*3\r\n$5\r\nPSYNC\r\n$1\r\n?\r\n$2\r\n-1\r\n";

            Socket masterSocket = new Socket(masterHostname, masterPortNumber);
            BufferedReader reader = new BufferedReader(new InputStreamReader(masterSocket.getInputStream()));
            DataOutputStream dos = new DataOutputStream(masterSocket.getOutputStream());
            dos.write(request_ping.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            String content = reader.readLine();
            System.out.println("ping->" + content);

            dos.write(request_replconf_1.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_replconf_1->" + content);

            dos.write(request_replconf_2.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_replconf_2->" + content);

            dos.write(request_PSYNC.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_PSYNC->" + content);

            BufferedWriter writer = new BufferedWriter(
                    new OutputStreamWriter(masterSocket.getOutputStream()));

            while ((content = reader.readLine()) != null) {
                System.out.println(" processMasterPropagation 读到的内容read content->::" + content);
                if ("GETACK".equalsIgnoreCase(content)) {
                    String toSend[] = {"REPLCONF", "ACK", "0"};
                    masterSocket.getOutputStream().write(encodeRESPArr(toSend).getBytes());
                } else if ("SET".equals(content)) {
                    System.out.println("processMasterPropagation 进入set命令 ,this is set");
                    Vector<String> command = new Vector<>();

                    reader.readLine();
                    String key = reader.readLine();
                    reader.readLine();
                    String value = reader.readLine();

                    command.addElement("SET");
                    command.addElement(key);
                    command.addElement(value);

                    sendToReplica(command);

                    currentKey = key;
                    System.out.println("processMasterPropagation BEGIN....SET");
                    map.put(key, value);
                    expMap.put(key, new KeyExpiry(0, true, Long.MAX_VALUE));
                    System.out.println("processMasterPropagation END.....SET");

                    System.out.println("this is a message from master propagation");

                } else if ("px".equalsIgnoreCase(content)) {
                    reader.readLine();
                    long exp = Long.valueOf(reader.readLine());
                    expMap.put(currentKey,
                            new KeyExpiry(exp, true, System.currentTimeMillis()));
                } else if ("GET".equals(content)) {
                    System.out.println("processMasterPropagation 进入GET命令 this is get");
                    //Thread.sleep(50);
                    reader.readLine();
                    content = reader.readLine();
                    System.out.println("processMasterPropagation BEGIN...... GET");
                    KeyExpiry keyExpiry = expMap.get(content);
                    System.out.println("processMasterPropagation END ..... ..GET");
                    String value = null;
                    System.out.println(keyExpiry.isActive);
                    if (keyExpiry.isActive) {
                        long currTime = System.currentTimeMillis();
                        long duration = currTime - keyExpiry.startTime;
                        if (duration < keyExpiry.expTime) {
                            value = map.get(content);
                        } else {
                            keyExpiry.isActive = false;
                        }
                    }
                    if (value == null) {
                        writer.write("$-1\r\n");
                        writer.flush();
                        continue;
                    }
                    StringBuilder sb = new StringBuilder();
                    sb.append("$" + value.length());
                    sb.append("\r\n").append(value).append("\r\n");
                    System.out.println(sb);
                    writer.write(sb.toString());
                    writer.flush();
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private static void processClients(Socket clientSocket) {

        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(clientSocket.getInputStream()));
             BufferedWriter writer = new BufferedWriter(
                     new OutputStreamWriter(clientSocket.getOutputStream()));) {
            String content;
            while ((content = reader.readLine()) != null) {
                System.out.println("processClients 读到的内容read content->::" + content);
                if ("ping".equalsIgnoreCase(content)) {
                    writer.write("+PONG\r\n");
                    writer.flush();
                } else if ("REPLCONF".equalsIgnoreCase(content)) {
                    writer.write("+OK\r\n");
                    writer.flush();
                } else if ("GETACK".equalsIgnoreCase(content)) {
                    String toSend[] = {"REPLCONF", "ACK", "0"};
                    clientSocket.getOutputStream().write(encodeRESPArr(toSend).getBytes());
                } else if ("PSYNC".equalsIgnoreCase(content)) {

                    String replid = "8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb";
                    String resp = "+FULLRESYNC " + replid + " 0\r\n";
                    writer.write(resp);
                    writer.flush();

                    byte[] contents = HexFormat.of().parseHex("524544495330303131fa0972656469732d76657205372e322e30fa0a72656469732d62697473c040fa056374696d65c26d08bc65fa08757365642d6d656dc2b0c41000fa08616f662d62617365c000fff06e3bfec0ff5aa2");
                    resp = "$" + contents.length + "\r\n";
                    clientSocket.getOutputStream().write(resp.getBytes());
                    clientSocket.getOutputStream().write(contents);

                    replicaSockets.add(clientSocket);

                } else if ("eof".equalsIgnoreCase(content)) {
                    System.out.println("eof");
                } else if ("ECHO".equalsIgnoreCase(content)) {
                    reader.readLine(); // Read and ignore the next line (length of the message)
                    String message = reader.readLine(); // Read the actual message
                    clientSocket.getOutputStream().write(
                            String.format("$%d\r\n%s\r\n", message.length(), message)
                                    .getBytes()); // Send the formatted message as a RESP bulk string
                } else if ("SET".equals(content)) {
                    System.out.println("processClients 进入set命令 ,this is set");
                    Vector<String> command = new Vector<>();

                    reader.readLine();
                    String key = reader.readLine();
                    reader.readLine();
                    String value = reader.readLine();

                    command.addElement("SET");
                    command.addElement(key);
                    command.addElement(value);

                    sendToReplica(command);

                    currentKey = key;
                    System.out.println("processClients BEGIN....SET");
                    map.put(key, value);
                    expMap.put(key, new KeyExpiry(0, true, Long.MAX_VALUE));
                    System.out.println("processClients END.....SET");

                    writer.write("$2\r\nOK\r\n");
                    writer.flush();

                } else if ("px".equalsIgnoreCase(content)) {
                    reader.readLine();
                    long exp = Long.valueOf(reader.readLine());
                    expMap.put(currentKey,
                            new KeyExpiry(exp, true, System.currentTimeMillis()));
                } else if ("GET".equals(content)) {
                    System.out.println("processClients 进入GET命令 this is get");
                    //Thread.sleep(50); // if i uncomment this , then it will pass the test, but i think it's not a proper solution.
                    reader.readLine();
                    content = reader.readLine();
                    System.out.println("processClients BEGIN...... GET");
                    KeyExpiry keyExpiry = expMap.get(content);
                    System.out.println("processClients END ..... ..GET");
                    String value = null;
                    System.out.println(keyExpiry.isActive);
                    if (keyExpiry.isActive) {
                        long currTime = System.currentTimeMillis();
                        long duration = currTime - keyExpiry.startTime;
                        if (duration < keyExpiry.expTime) {
                            value = map.get(content);
                        } else {
                            keyExpiry.isActive = false;
                        }
                    }
                    if (value == null) {
                        writer.write("$-1\r\n");
                        writer.flush();
                        continue;
                    }
                    StringBuilder sb = new StringBuilder();
                    sb.append("$" + value.length());
                    sb.append("\r\n").append(value).append("\r\n");
                    System.out.println(sb);
                    writer.write(sb.toString());
                    writer.flush();
                } else if ("info".equalsIgnoreCase(content)) {
                    System.out.println("info->" + content);
                    StringBuilder sb = new StringBuilder();
                    if (isMaster) {
                        sb.append("role:master\n").append("master_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\n").append("master_repl_offset:0");
                    } else {
                        sb.append("role:slave\n").append("master_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\n").append("master_repl_offset:0");
                    }
                    writer.write("$" + sb.toString().length() + "\r\n" + sb + "\r\n");
                    writer.flush();
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            try {
                if (clientSocket != null) {
                    clientSocket.close();
                }
            } catch (IOException e) {
                System.out.println("IOException: " + e.getMessage());
            }
        }
    }
}

```

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 13, 2025, 6:13am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/3 "2025-03-13T06:13:45Z")

</div>

Hey @show1999, there might be multiple issues at play. The first thing I noticed is that the handshake is being handled in a thread:

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/e/e07c67a4e8ce04700700e25b1a36287d39bf5214.png)

This isn’t ideal because the replica shouldn’t accept client connections until the handshake with the master is complete. In other words, **the handshake should be blocking until it finishes**.

Could you try moving the handshake out of the thread? Then we can proceed to address other potential issues.

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 13, 2025, 7:35am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/4 "2025-03-13T07:35:33Z")

</div>

OK, yes, thank you ,you’re right. I forgot about that point. Now I have moved the handshake out, but actually, before this, my handshake was placed outside this thread, and then there would be reception issues causing XV6 to fail. However, it’s not reproducing now, which is strange.  
Alright, I have now moved the handshake out, but YG4 still has the same issue—it can’t get the value. (Moreover, I noticed that in the YG4 phase, the replica did not read the RDB file sent by the master, which might be the reception issue I mentioned. It didn’t occur in the XV6 phase, but it appeared in the YG4 phase.)  
So, I have two questions I need your help with:

1. Regarding the replica receiving data sent by the master, it seems unstable. For example, this time in the YG4 phase, it didn’t receive the RDB file.
2. Regarding the issue of the `get` command executing before the `set` command, resulting in not getting the value.

I appreciate your help very much.

here is my log

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/8/82c85f48601963e45243829bbbd3d88ffc0befe7.jpeg)  
 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/5/5ba2e50fc3e399878c751e9561ee24a3d2fc2710.png)  
 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/a/a080b6d727bcb180dec3f1358ffd656fadef7e7e.jpeg)  
 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/3/36e0537a45fcbc1fba210f1a4b94abf3bcc0e966.png)  
 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/6/6fd501b19ec16300a04bc34499496fff997b4cba.png)

here is my code ，moving handshake out

```Java
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

class KeyExpiry {
    public long expTime;
    public boolean isActive;
    public long startTime;

    public KeyExpiry(long expTime, boolean isActive, long startTime) {
        this.expTime = expTime;
        this.isActive = isActive;
        this.startTime = startTime;
    }
}

public class Main {
    private static HashSet<Socket> replicaSockets = new HashSet<>();
    private static Map<String, String> map;
    private static Map<String, KeyExpiry> expMap;
    private static String currentKey = "";
    private static int port = 6379;
    private static boolean isMaster = true;

    public static void main(String[] args) {

        map = new ConcurrentHashMap<>();
        expMap = new ConcurrentHashMap<>();
        // You can use print statements as follows for debugging, they'll be visible when running tests.
        System.out.println("Logs from your program will appear here!");

        // Uncomment this block to pass the first stage
        ServerSocket serverSocket = null;

        handleArgs(List.of(args));
        try {
            serverSocket = new ServerSocket(port);

            // Since the tester restarts your program quite often, setting SO_REUSEADDR
            // ensures that we don't run into 'Address already in use' errors
            serverSocket.setReuseAddress(true);
            // Wait for connection from client.
            while (true) {
                Socket clientSocket = serverSocket.accept();

                new Thread(() -> {
                    try {
                        processClients(clientSocket);
                    } catch (Exception e) {
                        System.out.println("Exception: " + e.getMessage());
                    }
                }).start();
            }

        } catch (IOException e) {
            System.out.println("IOException: " + e.getMessage());
        }
    }

    private static void handleArgs(List<String> args) {
        if (args.contains("--port")) {
            port = Integer.parseInt(args.get(args.indexOf("--port") + 1));
        }
        if (args.contains("--replicaof")) {
            String[] replicaArgs = args.get(args.indexOf("--replicaof") + 1).split(" ");
            String masterHostname = replicaArgs[0];
            int masterPortNumber = Integer.parseInt(replicaArgs[1]);
            isMaster = false;
            replicaHandshake(masterHostname, masterPortNumber);
        }
    }

    private static void replicaHandshake(String masterHostname, int masterPortNumber) {
        try {
            String request_ping = "*1\r\n$4\r\nping\r\n";
            String request_replconf_1 = "*3\r\n$8\r\nREPLCONF\r\n$14\r\nlistening-port\r\n$4\r\n6380\r\n";
            String request_replconf_2 = "*3\r\n$8\r\nREPLCONF\r\n$4\r\ncapa\r\n$6\r\npsync2\r\n";
            String request_PSYNC = "*3\r\n$5\r\nPSYNC\r\n$1\r\n?\r\n$2\r\n-1\r\n";

            Socket masterSocket = new Socket(masterHostname, masterPortNumber);
            BufferedReader reader = new BufferedReader(new InputStreamReader(masterSocket.getInputStream()));
            DataOutputStream dos = new DataOutputStream(masterSocket.getOutputStream());

            dos.write(request_ping.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            String content = reader.readLine();
            System.out.println("ping->" + content);

            dos.write(request_replconf_1.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_replconf_1->" + content);

            dos.write(request_replconf_2.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_replconf_2->" + content);

            dos.write(request_PSYNC.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_PSYNC->" + content);

            new Thread(() -> {
                try {
                    processMasterPropagation(masterSocket);
                } catch (Exception e) {
                    System.out.println("Exception: " + e.getMessage());
                }
            }).start();

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static void sendToReplica(Vector<String> command) throws IOException {
        String toSend = "";
        String arr[] = new String[command.size()];
        for (int i = 0; i < command.size(); i++) {
            arr[i] = command.get(i);
        }
        toSend = encodeRESPArr(arr);
        for (Socket s : replicaSockets) {
            OutputStream os = s.getOutputStream();
            System.out.println("Writing to a replica");
            os.write(toSend.getBytes());
        }
    }

    public static String encodeRESPArr(String[] arr) {
        int n = arr.length;
        String send = "*" + n + "\r\n";
        for (String s : arr) {
            send += encodeRESP(s);
        }
        return send;
    }

    public static String encodeRESP(String s) {
        int size = s.length();
        String ret = "$" + size + "\r\n" + s + "\r\n";
        return ret;
    }

    private static void processMasterPropagation(Socket masterSocket) {

        try {
            BufferedReader reader = new BufferedReader(new InputStreamReader(masterSocket.getInputStream()));
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(masterSocket.getOutputStream()));
            String content;
            while ((content = reader.readLine()) != null) {
                System.out.println(" processMasterPropagation 读到的内容read content->::" + content);
                if ("GETACK".equalsIgnoreCase(content)) {
                    String toSend[] = {"REPLCONF", "ACK", "0"};
                    masterSocket.getOutputStream().write(encodeRESPArr(toSend).getBytes());
                } else if ("SET".equals(content)) {
                    System.out.println("processMasterPropagation 进入set命令 ,this is set");
                    Vector<String> command = new Vector<>();

                    reader.readLine();
                    String key = reader.readLine();
                    reader.readLine();
                    String value = reader.readLine();

                    command.addElement("SET");
                    command.addElement(key);
                    command.addElement(value);

                    sendToReplica(command);

                    currentKey = key;
                    System.out.println("processMasterPropagation BEGIN....SET");
                    map.put(key, value);
                    expMap.put(key, new KeyExpiry(0, true, Long.MAX_VALUE));
                    System.out.println("processMasterPropagation END.....SET");

                    System.out.println("this is a message from master propagation");

                } else if ("px".equalsIgnoreCase(content)) {
                    reader.readLine();
                    long exp = Long.valueOf(reader.readLine());
                    expMap.put(currentKey,
                            new KeyExpiry(exp, true, System.currentTimeMillis()));
                } else if ("GET".equals(content)) {
                    System.out.println("processMasterPropagation 进入GET命令 this is get");
                    //Thread.sleep(50);
                    reader.readLine();
                    content = reader.readLine();
                    System.out.println("processMasterPropagation BEGIN...... GET");
                    KeyExpiry keyExpiry = expMap.get(content);
                    System.out.println("processMasterPropagation END ..... ..GET");
                    String value = null;
                    System.out.println(keyExpiry.isActive);
                    if (keyExpiry.isActive) {
                        long currTime = System.currentTimeMillis();
                        long duration = currTime - keyExpiry.startTime;
                        if (duration < keyExpiry.expTime) {
                            value = map.get(content);
                        } else {
                            keyExpiry.isActive = false;
                        }
                    }
                    if (value == null) {
                        writer.write("$-1\r\n");
                        writer.flush();
                        continue;
                    }
                    StringBuilder sb = new StringBuilder();
                    sb.append("$" + value.length());
                    sb.append("\r\n").append(value).append("\r\n");
                    System.out.println(sb);
                    writer.write(sb.toString());
                    writer.flush();
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private static void processClients(Socket clientSocket) {

        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(clientSocket.getInputStream()));
             BufferedWriter writer = new BufferedWriter(
                     new OutputStreamWriter(clientSocket.getOutputStream()));) {
            String content;
            while ((content = reader.readLine()) != null) {
                System.out.println("processClients 读到的内容read content->::" + content);
                if ("ping".equalsIgnoreCase(content)) {
                    writer.write("+PONG\r\n");
                    writer.flush();
                } else if ("REPLCONF".equalsIgnoreCase(content)) {
                    writer.write("+OK\r\n");
                    writer.flush();
                } else if ("GETACK".equalsIgnoreCase(content)) {
                    String toSend[] = {"REPLCONF", "ACK", "0"};
                    clientSocket.getOutputStream().write(encodeRESPArr(toSend).getBytes());
                } else if ("PSYNC".equalsIgnoreCase(content)) {

                    String replid = "8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb";
                    String resp = "+FULLRESYNC " + replid + " 0\r\n";
                    writer.write(resp);
                    writer.flush();

                    byte[] contents = HexFormat.of().parseHex("524544495330303131fa0972656469732d76657205372e322e30fa0a72656469732d62697473c040fa056374696d65c26d08bc65fa08757365642d6d656dc2b0c41000fa08616f662d62617365c000fff06e3bfec0ff5aa2");
                    resp = "$" + contents.length + "\r\n";
                    clientSocket.getOutputStream().write(resp.getBytes());
                    clientSocket.getOutputStream().write(contents);

                    replicaSockets.add(clientSocket);

                } else if ("eof".equalsIgnoreCase(content)) {
                    System.out.println("eof");
                } else if ("ECHO".equalsIgnoreCase(content)) {
                    reader.readLine(); // Read and ignore the next line (length of the message)
                    String message = reader.readLine(); // Read the actual message
                    clientSocket.getOutputStream().write(
                            String.format("$%d\r\n%s\r\n", message.length(), message)
                                    .getBytes()); // Send the formatted message as a RESP bulk string
                } else if ("SET".equals(content)) {
                    System.out.println("processClients 进入set命令 ,this is set");
                    Vector<String> command = new Vector<>();

                    reader.readLine();
                    String key = reader.readLine();
                    reader.readLine();
                    String value = reader.readLine();

                    command.addElement("SET");
                    command.addElement(key);
                    command.addElement(value);

                    sendToReplica(command);

                    currentKey = key;
                    System.out.println("processClients BEGIN....SET");
                    map.put(key, value);
                    expMap.put(key, new KeyExpiry(0, true, Long.MAX_VALUE));
                    System.out.println("processClients END.....SET");

                    writer.write("$2\r\nOK\r\n");
                    writer.flush();

                } else if ("px".equalsIgnoreCase(content)) {
                    reader.readLine();
                    long exp = Long.valueOf(reader.readLine());
                    expMap.put(currentKey,
                            new KeyExpiry(exp, true, System.currentTimeMillis()));
                } else if ("GET".equals(content)) {
                    System.out.println("processClients 进入GET命令 this is get");
                    //Thread.sleep(50); // if i uncomment this , then it will pass the test, but i think it's not a proper solution.
                    reader.readLine();
                    content = reader.readLine();
                    System.out.println("processClients BEGIN...... GET");
                    KeyExpiry keyExpiry = expMap.get(content);
                    System.out.println("processClients END ..... ..GET");
                    String value = null;
                    System.out.println(keyExpiry.isActive);
                    if (keyExpiry.isActive) {
                        long currTime = System.currentTimeMillis();
                        long duration = currTime - keyExpiry.startTime;
                        if (duration < keyExpiry.expTime) {
                            value = map.get(content);
                        } else {
                            keyExpiry.isActive = false;
                        }
                    }
                    if (value == null) {
                        writer.write("$-1\r\n");
                        writer.flush();
                        continue;
                    }
                    StringBuilder sb = new StringBuilder();
                    sb.append("$" + value.length());
                    sb.append("\r\n").append(value).append("\r\n");
                    System.out.println(sb);
                    writer.write(sb.toString());
                    writer.flush();
                } else if ("info".equalsIgnoreCase(content)) {
                    System.out.println("info->" + content);
                    StringBuilder sb = new StringBuilder();
                    if (isMaster) {
                        sb.append("role:master\n").append("master_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\n").append("master_repl_offset:0");
                    } else {
                        sb.append("role:slave\n").append("master_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\n").append("master_repl_offset:0");
                    }
                    writer.write("$" + sb.toString().length() + "\r\n" + sb + "\r\n");
                    writer.flush();
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            try {
                if (clientSocket != null) {
                    clientSocket.close();
                }
            } catch (IOException e) {
                System.out.println("IOException: " + e.getMessage());
            }
        }
    }
}

```

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 13, 2025, 7:59am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/5 "2025-03-13T07:59:28Z")

</div>

> 1. Regarding the replica receiving data sent by the master, it seems unstable. For example, this time in the YG4 phase, it didn’t receive the RDB file.

Could you point me to where you’re handling RDB reading?

Please note that while RDB may sometimes be read along with PSYNC, this isn’t guaranteed and shouldn’t be relied upon.

> 1. Regarding the issue of the `get` command executing before the `set` command, resulting in not getting the value.

Let’s address this after ensuring RDB is handled properly.

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 13, 2025, 8:37am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/6 "2025-03-13T08:37:50Z")

</div>

> [@andy1li](#):
>
> Could you point me to where you’re handling RDB reading?
> 
> Please note that while RDB may sometimes be read along with PSYNC, this isn’t guaranteed and shouldn’t be relied upon.

I believe this is where the RDB file is being read. As you can see, in the XV6 phase, the RDB file was successfully read, but in the YG4 phase, it was not read.

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/7/77f5f53ca48ca563f68bdb9b195ddeb6ef823e13.png)  
Am I misunderstanding? Could you please explain it in more detail? Thank you.

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 13, 2025, 10:01am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/7 "2025-03-13T10:01:56Z")

</div>

Hmm, I see. It seems that you’re implicitly handling and ignoring RDB there.

You might want to make sure RDB is properly handled/ignored **before** starting the thread for `processMasterPropagation`, as RDB is part of the handshake/resynchronization process.

* * *

I also noticed is that two `BufferedReader`s are being created from masterSocket. This could cause RDB data to be buffered in the first one, potentially leading to instability in RDB handling.

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/7/71097f3b8688092b836f5c8bf38685a2441e5deb.jpeg)

There may be other issues as well, but let’s tackle them as we go.

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 13, 2025, 11:53am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/8 "2025-03-13T11:53:48Z")

</div>

ok，love you， so question 1 is solved，  
help me about question2 ， appreciate your help very much @andy1li

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 13, 2025, 11:58am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/10 "2025-03-13T11:58:51Z")

</div>

@show1999 Feel free to tag me once you’ve addressed this:

> You might want to make sure RDB is properly handled/ignored **before** starting the thread for `processMasterPropagation` , as RDB is part of the handshake/resynchronization process.

When that’s done, I’ll take another look at your code.

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 13, 2025, 12:05pm UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/11 "2025-03-13T12:05:45Z")

</div>

@andy1li  
yes，i have already solved the question1 with your help， just keep one reader.

here is my new code

```Java
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

class KeyExpiry {
    public long expTime;
    public boolean isActive;
    public long startTime;

    public KeyExpiry(long expTime, boolean isActive, long startTime) {
        this.expTime = expTime;
        this.isActive = isActive;
        this.startTime = startTime;
    }
}

public class Main {
    private static HashSet<Socket> replicaSockets = new HashSet<>();
    private static Map<String, String> map;
    private static Map<String, KeyExpiry> expMap;
    private static String currentKey = "";
    private static int port = 6379;
    private static boolean isMaster = true;

    public static void main(String[] args) {

        map = new ConcurrentHashMap<>();
        expMap = new ConcurrentHashMap<>();
        // You can use print statements as follows for debugging, they'll be visible when running tests.
        System.out.println("Logs from your program will appear here!");

        // Uncomment this block to pass the first stage
        ServerSocket serverSocket = null;

        handleArgs(List.of(args));
        try {
            serverSocket = new ServerSocket(port);

            // Since the tester restarts your program quite often, setting SO_REUSEADDR
            // ensures that we don't run into 'Address already in use' errors
            serverSocket.setReuseAddress(true);
            // Wait for connection from client.
            while (true) {
                Socket clientSocket = serverSocket.accept();

                new Thread(() -> {
                    try {
                        processClients(clientSocket);
                    } catch (Exception e) {
                        System.out.println("Exception: " + e.getMessage());
                    }
                }).start();
            }

        } catch (IOException e) {
            System.out.println("IOException: " + e.getMessage());
        }
    }

    private static void handleArgs(List<String> args) {
        if (args.contains("--port")) {
            port = Integer.parseInt(args.get(args.indexOf("--port") + 1));
        }
        if (args.contains("--replicaof")) {
            String[] replicaArgs = args.get(args.indexOf("--replicaof") + 1).split(" ");
            String masterHostname = replicaArgs[0];
            int masterPortNumber = Integer.parseInt(replicaArgs[1]);
            isMaster = false;
            replicaHandshake(masterHostname, masterPortNumber);
        }
    }

    private static void replicaHandshake(String masterHostname, int masterPortNumber) {
        try {
            String request_ping = "*1\r\n$4\r\nping\r\n";
            String request_replconf_1 = "*3\r\n$8\r\nREPLCONF\r\n$14\r\nlistening-port\r\n$4\r\n6380\r\n";
            String request_replconf_2 = "*3\r\n$8\r\nREPLCONF\r\n$4\r\ncapa\r\n$6\r\npsync2\r\n";
            String request_PSYNC = "*3\r\n$5\r\nPSYNC\r\n$1\r\n?\r\n$2\r\n-1\r\n";

            Socket masterSocket = new Socket(masterHostname, masterPortNumber);
            BufferedReader reader = new BufferedReader(new InputStreamReader(masterSocket.getInputStream()));
            DataOutputStream dos = new DataOutputStream(masterSocket.getOutputStream());

            dos.write(request_ping.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            String content = reader.readLine();
            System.out.println("ping->" + content);

            dos.write(request_replconf_1.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_replconf_1->" + content);

            dos.write(request_replconf_2.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_replconf_2->" + content);

            dos.write(request_PSYNC.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_PSYNC->" + content);

            new Thread(() -> {
                try {
                    processMasterPropagation(masterSocket,reader);
                } catch (Exception e) {
                    System.out.println("Exception: " + e.getMessage());
                }
            }).start();

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static void sendToReplica(Vector<String> command) throws IOException {
        String toSend = "";
        String arr[] = new String[command.size()];
        for (int i = 0; i < command.size(); i++) {
            arr[i] = command.get(i);
        }
        toSend = encodeRESPArr(arr);
        for (Socket s : replicaSockets) {
            OutputStream os = s.getOutputStream();
            System.out.println("Writing to a replica");
            os.write(toSend.getBytes());
        }
    }

    public static String encodeRESPArr(String[] arr) {
        int n = arr.length;
        String send = "*" + n + "\r\n";
        for (String s : arr) {
            send += encodeRESP(s);
        }
        return send;
    }

    public static String encodeRESP(String s) {
        int size = s.length();
        String ret = "$" + size + "\r\n" + s + "\r\n";
        return ret;
    }

    private static void processMasterPropagation(Socket masterSocket,BufferedReader reader) {

        try {
// BufferedReader reader = new BufferedReader(new InputStreamReader(masterSocket.getInputStream()));
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(masterSocket.getOutputStream()));
            String content;
            while ((content = reader.readLine()) != null) {
                System.out.println(" processMasterPropagation 读到的内容read content->::" + content);
                if ("GETACK".equalsIgnoreCase(content)) {
                    String toSend[] = {"REPLCONF", "ACK", "0"};
                    masterSocket.getOutputStream().write(encodeRESPArr(toSend).getBytes());
                } else if ("SET".equals(content)) {
                    System.out.println("processMasterPropagation 进入set命令 ,this is set");
                    Vector<String> command = new Vector<>();

                    reader.readLine();
                    String key = reader.readLine();
                    reader.readLine();
                    String value = reader.readLine();

                    command.addElement("SET");
                    command.addElement(key);
                    command.addElement(value);

                    sendToReplica(command);

                    currentKey = key;
                    System.out.println("processMasterPropagation BEGIN....SET");
                    map.put(key, value);
                    expMap.put(key, new KeyExpiry(0, true, Long.MAX_VALUE));
                    System.out.println("processMasterPropagation END.....SET");

                    System.out.println("this is a message from master propagation");

                } else if ("px".equalsIgnoreCase(content)) {
                    reader.readLine();
                    long exp = Long.valueOf(reader.readLine());
                    expMap.put(currentKey,
                            new KeyExpiry(exp, true, System.currentTimeMillis()));
                } else if ("GET".equals(content)) {
                    System.out.println("processMasterPropagation 进入GET命令 this is get");
                    //Thread.sleep(50);
                    reader.readLine();
                    content = reader.readLine();
                    System.out.println("processMasterPropagation BEGIN...... GET");
                    KeyExpiry keyExpiry = expMap.get(content);
                    System.out.println("processMasterPropagation END ..... ..GET");
                    String value = null;
                    System.out.println(keyExpiry.isActive);
                    if (keyExpiry.isActive) {
                        long currTime = System.currentTimeMillis();
                        long duration = currTime - keyExpiry.startTime;
                        if (duration < keyExpiry.expTime) {
                            value = map.get(content);
                        } else {
                            keyExpiry.isActive = false;
                        }
                    }
                    if (value == null) {
                        writer.write("$-1\r\n");
                        writer.flush();
                        continue;
                    }
                    StringBuilder sb = new StringBuilder();
                    sb.append("$" + value.length());
                    sb.append("\r\n").append(value).append("\r\n");
                    System.out.println(sb);
                    writer.write(sb.toString());
                    writer.flush();
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private static void processClients(Socket clientSocket) {

        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(clientSocket.getInputStream()));
             BufferedWriter writer = new BufferedWriter(
                     new OutputStreamWriter(clientSocket.getOutputStream()));) {
            String content;
            while ((content = reader.readLine()) != null) {
                System.out.println("processClients 读到的内容read content->::" + content);
                if ("ping".equalsIgnoreCase(content)) {
                    writer.write("+PONG\r\n");
                    writer.flush();
                } else if ("REPLCONF".equalsIgnoreCase(content)) {
                    writer.write("+OK\r\n");
                    writer.flush();
                } else if ("GETACK".equalsIgnoreCase(content)) {
                    String toSend[] = {"REPLCONF", "ACK", "0"};
                    clientSocket.getOutputStream().write(encodeRESPArr(toSend).getBytes());
                } else if ("PSYNC".equalsIgnoreCase(content)) {

                    String replid = "8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb";
                    String resp = "+FULLRESYNC " + replid + " 0\r\n";
                    writer.write(resp);
                    writer.flush();

                    byte[] contents = HexFormat.of().parseHex("524544495330303131fa0972656469732d76657205372e322e30fa0a72656469732d62697473c040fa056374696d65c26d08bc65fa08757365642d6d656dc2b0c41000fa08616f662d62617365c000fff06e3bfec0ff5aa2");
                    resp = "$" + contents.length + "\r\n";
                    clientSocket.getOutputStream().write(resp.getBytes());
                    clientSocket.getOutputStream().write(contents);

                    replicaSockets.add(clientSocket);

                } else if ("eof".equalsIgnoreCase(content)) {
                    System.out.println("eof");
                } else if ("ECHO".equalsIgnoreCase(content)) {
                    reader.readLine(); // Read and ignore the next line (length of the message)
                    String message = reader.readLine(); // Read the actual message
                    clientSocket.getOutputStream().write(
                            String.format("$%d\r\n%s\r\n", message.length(), message)
                                    .getBytes()); // Send the formatted message as a RESP bulk string
                } else if ("SET".equals(content)) {
                    System.out.println("processClients 进入set命令 ,this is set");
                    Vector<String> command = new Vector<>();

                    reader.readLine();
                    String key = reader.readLine();
                    reader.readLine();
                    String value = reader.readLine();

                    command.addElement("SET");
                    command.addElement(key);
                    command.addElement(value);

                    sendToReplica(command);

                    currentKey = key;
                    System.out.println("processClients BEGIN....SET");
                    map.put(key, value);
                    expMap.put(key, new KeyExpiry(0, true, Long.MAX_VALUE));
                    System.out.println("processClients END.....SET");

                    writer.write("$2\r\nOK\r\n");
                    writer.flush();

                } else if ("px".equalsIgnoreCase(content)) {
                    reader.readLine();
                    long exp = Long.valueOf(reader.readLine());
                    expMap.put(currentKey,
                            new KeyExpiry(exp, true, System.currentTimeMillis()));
                } else if ("GET".equals(content)) {
                    System.out.println("processClients 进入GET命令 this is get");
                    //Thread.sleep(50); // if i uncomment this , then it will pass the test, but i think it's not a proper solution.
                    reader.readLine();
                    content = reader.readLine();
                    System.out.println("processClients BEGIN...... GET");
                    KeyExpiry keyExpiry = expMap.get(content);
                    System.out.println("processClients END ..... ..GET");
                    String value = null;
                    System.out.println(keyExpiry.isActive);
                    if (keyExpiry.isActive) {
                        long currTime = System.currentTimeMillis();
                        long duration = currTime - keyExpiry.startTime;
                        if (duration < keyExpiry.expTime) {
                            value = map.get(content);
                        } else {
                            keyExpiry.isActive = false;
                        }
                    }
                    if (value == null) {
                        writer.write("$-1\r\n");
                        writer.flush();
                        continue;
                    }
                    StringBuilder sb = new StringBuilder();
                    sb.append("$" + value.length());
                    sb.append("\r\n").append(value).append("\r\n");
                    System.out.println(sb);
                    writer.write(sb.toString());
                    writer.flush();
                } else if ("info".equalsIgnoreCase(content)) {
                    System.out.println("info->" + content);
                    StringBuilder sb = new StringBuilder();
                    if (isMaster) {
                        sb.append("role:master\n").append("master_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\n").append("master_repl_offset:0");
                    } else {
                        sb.append("role:slave\n").append("master_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\n").append("master_repl_offset:0");
                    }
                    writer.write("$" + sb.toString().length() + "\r\n" + sb + "\r\n");
                    writer.flush();
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            try {
                if (clientSocket != null) {
                    clientSocket.close();
                }
            } catch (IOException e) {
                System.out.println("IOException: " + e.getMessage());
            }
        }
    }
}

```

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 13, 2025, 12:10pm UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/12 "2025-03-13T12:10:46Z")

</div>

@show1999 Good job using only one reader!

Your latest submission still shows that RDB is being handled **inside** the thread rather than **before** starting it:

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/9/9921aecf270aac605f24928f04feb628e0e75c96.png)

This may or may not be affecting **Issue 2** , but until it’s fixed, we can’t be certain. 😅

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 13, 2025, 12:26pm UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/13 "2025-03-13T12:26:36Z")

</div>

@andy1li

sorry i misunderstood your meaning before， so now I really receive RDB file before start a new thread.

it appears a new problem, see my log please.

you can see my latest code and log ,right?

anyway i post my code and log here,

here is my log

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/d/d7ee3d630fddb3a25d6a83c4d7822327ea6bda42.png)

here is my new code

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/b/bb82f89b22ca5f73b75117cf4601f2eeea816fea.png)

```java
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

class KeyExpiry {
    public long expTime;
    public boolean isActive;
    public long startTime;

    public KeyExpiry(long expTime, boolean isActive, long startTime) {
        this.expTime = expTime;
        this.isActive = isActive;
        this.startTime = startTime;
    }
}

public class Main {
    private static HashSet<Socket> replicaSockets = new HashSet<>();
    private static Map<String, String> map;
    private static Map<String, KeyExpiry> expMap;
    private static String currentKey = "";
    private static int port = 6379;
    private static boolean isMaster = true;

    public static void main(String[] args) {

        map = new ConcurrentHashMap<>();
        expMap = new ConcurrentHashMap<>();
        // You can use print statements as follows for debugging, they'll be visible when running tests.
        System.out.println("Logs from your program will appear here!");

        // Uncomment this block to pass the first stage
        ServerSocket serverSocket = null;

        handleArgs(List.of(args));
        try {
            serverSocket = new ServerSocket(port);

            // Since the tester restarts your program quite often, setting SO_REUSEADDR
            // ensures that we don't run into 'Address already in use' errors
            serverSocket.setReuseAddress(true);
            // Wait for connection from client.
            while (true) {
                Socket clientSocket = serverSocket.accept();

                new Thread(() -> {
                    try {
                        processClients(clientSocket);
                    } catch (Exception e) {
                        System.out.println("Exception: " + e.getMessage());
                    }
                }).start();
            }

        } catch (IOException e) {
            System.out.println("IOException: " + e.getMessage());
        }
    }

    private static void handleArgs(List<String> args) {
        if (args.contains("--port")) {
            port = Integer.parseInt(args.get(args.indexOf("--port") + 1));
        }
        if (args.contains("--replicaof")) {
            String[] replicaArgs = args.get(args.indexOf("--replicaof") + 1).split(" ");
            String masterHostname = replicaArgs[0];
            int masterPortNumber = Integer.parseInt(replicaArgs[1]);
            isMaster = false;
            replicaHandshake(masterHostname, masterPortNumber);
        }
    }

    private static void replicaHandshake(String masterHostname, int masterPortNumber) {
        try {
            String request_ping = "*1\r\n$4\r\nping\r\n";
            String request_replconf_1 = "*3\r\n$8\r\nREPLCONF\r\n$14\r\nlistening-port\r\n$4\r\n6380\r\n";
            String request_replconf_2 = "*3\r\n$8\r\nREPLCONF\r\n$4\r\ncapa\r\n$6\r\npsync2\r\n";
            String request_PSYNC = "*3\r\n$5\r\nPSYNC\r\n$1\r\n?\r\n$2\r\n-1\r\n";

            Socket masterSocket = new Socket(masterHostname, masterPortNumber);
            BufferedReader reader = new BufferedReader(new InputStreamReader(masterSocket.getInputStream()));
            DataOutputStream dos = new DataOutputStream(masterSocket.getOutputStream());

            dos.write(request_ping.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            String content = reader.readLine();
            System.out.println("ping->" + content);

            dos.write(request_replconf_1.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_replconf_1->" + content);

            dos.write(request_replconf_2.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_replconf_2->" + content);

            dos.write(request_PSYNC.getBytes(StandardCharsets.UTF_8));
            dos.flush();
            content = reader.readLine();
            System.out.println("request_PSYNC->" + content);

            content = reader.readLine();
            System.out.println("RDB----part1------>" + content);

            content = reader.readLine();
            System.out.println("RDB----part2------>" + content);

            content = reader.readLine();
            System.out.println("RDB----part3------>" + content);

            new Thread(() -> {
                try {
                    processMasterPropagation(masterSocket,reader);
                } catch (Exception e) {
                    System.out.println("Exception: " + e.getMessage());
                }
            }).start();

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static void sendToReplica(Vector<String> command) throws IOException {
        String toSend = "";
        String arr[] = new String[command.size()];
        for (int i = 0; i < command.size(); i++) {
            arr[i] = command.get(i);
        }
        toSend = encodeRESPArr(arr);
        for (Socket s : replicaSockets) {
            OutputStream os = s.getOutputStream();
            System.out.println("Writing to a replica");
            os.write(toSend.getBytes());
        }
    }

    public static String encodeRESPArr(String[] arr) {
        int n = arr.length;
        String send = "*" + n + "\r\n";
        for (String s : arr) {
            send += encodeRESP(s);
        }
        return send;
    }

    public static String encodeRESP(String s) {
        int size = s.length();
        String ret = "$" + size + "\r\n" + s + "\r\n";
        return ret;
    }

    private static void processMasterPropagation(Socket masterSocket,BufferedReader reader) {

        try {
// BufferedReader reader = new BufferedReader(new InputStreamReader(masterSocket.getInputStream()));
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(masterSocket.getOutputStream()));
            String content;
            while ((content = reader.readLine()) != null) {
                System.out.println(" processMasterPropagation 读到的内容read content->::" + content);
                if ("GETACK".equalsIgnoreCase(content)) {
                    String toSend[] = {"REPLCONF", "ACK", "0"};
                    masterSocket.getOutputStream().write(encodeRESPArr(toSend).getBytes());
                } else if ("SET".equals(content)) {
                    System.out.println("processMasterPropagation 进入set命令 ,this is set");
                    Vector<String> command = new Vector<>();

                    reader.readLine();
                    String key = reader.readLine();
                    reader.readLine();
                    String value = reader.readLine();

                    command.addElement("SET");
                    command.addElement(key);
                    command.addElement(value);

                    sendToReplica(command);

                    currentKey = key;
                    System.out.println("processMasterPropagation BEGIN....SET");
                    map.put(key, value);
                    expMap.put(key, new KeyExpiry(0, true, Long.MAX_VALUE));
                    System.out.println("processMasterPropagation END.....SET");

                    System.out.println("this is a message from master propagation");

                } else if ("px".equalsIgnoreCase(content)) {
                    reader.readLine();
                    long exp = Long.valueOf(reader.readLine());
                    expMap.put(currentKey,
                            new KeyExpiry(exp, true, System.currentTimeMillis()));
                } else if ("GET".equals(content)) {
                    System.out.println("processMasterPropagation 进入GET命令 this is get");
                    //Thread.sleep(50);
                    reader.readLine();
                    content = reader.readLine();
                    System.out.println("processMasterPropagation BEGIN...... GET");
                    KeyExpiry keyExpiry = expMap.get(content);
                    System.out.println("processMasterPropagation END ..... ..GET");
                    String value = null;
                    System.out.println(keyExpiry.isActive);
                    if (keyExpiry.isActive) {
                        long currTime = System.currentTimeMillis();
                        long duration = currTime - keyExpiry.startTime;
                        if (duration < keyExpiry.expTime) {
                            value = map.get(content);
                        } else {
                            keyExpiry.isActive = false;
                        }
                    }
                    if (value == null) {
                        writer.write("$-1\r\n");
                        writer.flush();
                        continue;
                    }
                    StringBuilder sb = new StringBuilder();
                    sb.append("$" + value.length());
                    sb.append("\r\n").append(value).append("\r\n");
                    System.out.println(sb);
                    writer.write(sb.toString());
                    writer.flush();
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private static void processClients(Socket clientSocket) {

        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(clientSocket.getInputStream()));
             BufferedWriter writer = new BufferedWriter(
                     new OutputStreamWriter(clientSocket.getOutputStream()));) {
            String content;
            while ((content = reader.readLine()) != null) {
                System.out.println("processClients 读到的内容read content->::" + content);
                if ("ping".equalsIgnoreCase(content)) {
                    writer.write("+PONG\r\n");
                    writer.flush();
                } else if ("REPLCONF".equalsIgnoreCase(content)) {
                    writer.write("+OK\r\n");
                    writer.flush();
                } else if ("GETACK".equalsIgnoreCase(content)) {
                    String toSend[] = {"REPLCONF", "ACK", "0"};
                    clientSocket.getOutputStream().write(encodeRESPArr(toSend).getBytes());
                } else if ("PSYNC".equalsIgnoreCase(content)) {

                    String replid = "8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb";
                    String resp = "+FULLRESYNC " + replid + " 0\r\n";
                    writer.write(resp);
                    writer.flush();

                    byte[] contents = HexFormat.of().parseHex("524544495330303131fa0972656469732d76657205372e322e30fa0a72656469732d62697473c040fa056374696d65c26d08bc65fa08757365642d6d656dc2b0c41000fa08616f662d62617365c000fff06e3bfec0ff5aa2");
                    resp = "$" + contents.length + "\r\n";
                    clientSocket.getOutputStream().write(resp.getBytes());
                    clientSocket.getOutputStream().write(contents);

                    replicaSockets.add(clientSocket);

                } else if ("eof".equalsIgnoreCase(content)) {
                    System.out.println("eof");
                } else if ("ECHO".equalsIgnoreCase(content)) {
                    reader.readLine(); // Read and ignore the next line (length of the message)
                    String message = reader.readLine(); // Read the actual message
                    clientSocket.getOutputStream().write(
                            String.format("$%d\r\n%s\r\n", message.length(), message)
                                    .getBytes()); // Send the formatted message as a RESP bulk string
                } else if ("SET".equals(content)) {
                    System.out.println("processClients 进入set命令 ,this is set");
                    Vector<String> command = new Vector<>();

                    reader.readLine();
                    String key = reader.readLine();
                    reader.readLine();
                    String value = reader.readLine();

                    command.addElement("SET");
                    command.addElement(key);
                    command.addElement(value);

                    sendToReplica(command);

                    currentKey = key;
                    System.out.println("processClients BEGIN....SET");
                    map.put(key, value);
                    expMap.put(key, new KeyExpiry(0, true, Long.MAX_VALUE));
                    System.out.println("processClients END.....SET");

                    writer.write("$2\r\nOK\r\n");
                    writer.flush();

                } else if ("px".equalsIgnoreCase(content)) {
                    reader.readLine();
                    long exp = Long.valueOf(reader.readLine());
                    expMap.put(currentKey,
                            new KeyExpiry(exp, true, System.currentTimeMillis()));
                } else if ("GET".equals(content)) {
                    System.out.println("processClients 进入GET命令 this is get");
                    //Thread.sleep(50); // if i uncomment this , then it will pass the test, but i think it's not a proper solution.
                    reader.readLine();
                    content = reader.readLine();
                    System.out.println("processClients BEGIN...... GET");
                    KeyExpiry keyExpiry = expMap.get(content);
                    System.out.println("processClients END ..... ..GET");
                    String value = null;
                    System.out.println(keyExpiry.isActive);
                    if (keyExpiry.isActive) {
                        long currTime = System.currentTimeMillis();
                        long duration = currTime - keyExpiry.startTime;
                        if (duration < keyExpiry.expTime) {
                            value = map.get(content);
                        } else {
                            keyExpiry.isActive = false;
                        }
                    }
                    if (value == null) {
                        writer.write("$-1\r\n");
                        writer.flush();
                        continue;
                    }
                    StringBuilder sb = new StringBuilder();
                    sb.append("$" + value.length());
                    sb.append("\r\n").append(value).append("\r\n");
                    System.out.println(sb);
                    writer.write(sb.toString());
                    writer.flush();
                } else if ("info".equalsIgnoreCase(content)) {
                    System.out.println("info->" + content);
                    StringBuilder sb = new StringBuilder();
                    if (isMaster) {
                        sb.append("role:master\n").append("master_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\n").append("master_repl_offset:0");
                    } else {
                        sb.append("role:slave\n").append("master_replid:8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb\n").append("master_repl_offset:0");
                    }
                    writer.write("$" + sb.toString().length() + "\r\n" + sb + "\r\n");
                    writer.flush();
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            try {
                if (clientSocket != null) {
                    clientSocket.close();
                }
            } catch (IOException e) {
                System.out.println("IOException: " + e.getMessage());
            }
        }
    }
}

```

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 14, 2025, 7:38am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/14 "2025-03-14T07:38:10Z")

</div>

I logged each character from `content` in the format: `char (hex | decimal)`.

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/c/c5b5c3612b783f5e0ac6a17a6d11728a2bb4a3cd.png)

You can see that the hex value `fa` is not being read correctly:

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/a/a4d65830e78a060877dec2c64b98c81bfd150fe7.jpeg)

The issue seems to be caused by reading data as characters instead of as raw bytes, because `\xfa` is not a valid UTF-8 character.

You might need to use **DataInputStream** instead of **BufferedReader** to correctly handle raw byte data.

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 14, 2025, 8:04am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/16 "2025-03-14T08:04:13Z")

</div>

@andy1li  
use DataInputStream instead of  
BufferedReader  
I don’t understand why this issue would affect the result, but I made the changes anyway. However, the outcome seems to remain the same. 🙃  
thank you for your replys ， appreciate it very much  
here is part of my log：

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/9/933c38a5e436a4ce4947e4de06e40be07e5053b2.png)  
and my code just change - → use DataInputStream instead of  
BufferedReader

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 14, 2025, 8:06am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/17 "2025-03-14T08:06:47Z")

</div>

Honestly, I’m not sure about the root cause yet, as there could be multiple factors involved. I can only identify them one at a time. 🤝

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 14, 2025, 8:08am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/18 "2025-03-14T08:08:39Z")

</div>

ok , you are right ,just fix it one by one

just help me please , appreciate it.

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 14, 2025, 8:17am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/19 "2025-03-14T08:17:31Z")

</div>

> master: Sent bytes: “$88\r\nREDIS0011\xfa\tredis-ver\x057.2.0\xfa\nredis-bits\xc0@\xfa\x05ctime\xc2m\b\xbce\xfa\bused-mem°\xc4\x10\x00\xfa\baof-base\xc0\x00\xff\xf0n;\xfe\xc0\xffZ\xa2”

The next issue is that the empty RDB file does not end with a **newline** , causing `readLine` to wait indefinitely for the newline that never arrives:

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/0/0c3363c1e8298534316cffabfa8842b9b4e18398.png)

To resolve this, read the exact number of bytes (88 in this case) instead of relying on `readLine`.

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 14, 2025, 9:19am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/20 "2025-03-14T09:19:07Z")

</div>

@andy1li

I’m not sure if I understand correctly. Are you saying that the main thread will block here and stop running further because the `readLine` method did not encounter a newline character? I tried it and found that it can continue to run.

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/e/e15fe7449d4c02615e023d5b84767e80bb050cc9.png)

Anyway, I still tried to change the `readLine` method to accurately read 88 bytes. I expected it to throw the same error (connection refused) to prove that it has nothing to do with `readLine` , but it actually didn’t encounter the connection refused error.  
This has left me very confused. Could you please explain this in more details?

Thank you. ❤

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 14, 2025, 9:23am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/21 "2025-03-14T09:23:28Z")

</div>

> Are you saying that the main thread will block here and stop running further because the `readLine` method did not encounter a newline character?

Yep, and that’s one of the issues we’re trying to resolve.

> I tried it and found that it can continue to run.

Could you share more details about your findings?

---

<div class="post-metadata">

**Author:** ![show1999](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/show1999/32/6637_2.png) [@show1999](https://forum.codecrafters.io/u/show1999)\
**Post date:** [March 14, 2025, 9:28am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/22 "2025-03-14T09:28:59Z")

</div>

@andy1li  
hi

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/9/924e4e9a9d40b0e59e2e136730ff20c9ea66b836.png)  
This is part of my log. It printed “yes,” indicating that the main thread can continue running and is not blocked by readLine.

am I right？ i dont know。 🙃

and also ， why can i pass xv6

i’m very confused.

---

<div class="post-metadata">

**Author:** ![andy1li](https://yyz1.discourse-cdn.com/flex003/user_avatar/forum.codecrafters.io/andy1li/32/10_2.png) [@andy1li](https://forum.codecrafters.io/u/andy1li)\
**Post date:** [March 14, 2025, 9:36am UTC](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610/23 "2025-03-14T09:36:21Z")

</div>

Ah, got it!

It was blocking until a client in the tester tried to connect to your replica, but the connection failed with `dial tcp [::1]:6380: connect: connection refused`:

 ![image](https://canada1.discourse-cdn.com/flex003/uploads/codecrafters/original/2X/e/ea60a1bc26165a0cf572e1bce772d2a5ba5d6f6b.png)

As a result, the tester tried to terminate your replica, causing it to stop blocking and proceed to print the buffered content for **RDB----Part 3**.

Does that make sense?

[Next page](https://forum.codecrafters.io/t/yg4-client-get-command-before-propagation-set-command-so-get/6610.md?page=2)