Redis ZN8: Replica is receiving propagated command with unnecessary padded zeros (Java)

[lakshya-k]

I’m stuck on Stage #ZN8: Single-replica propagation.

The replica is able to complete the handshake after which the master is supposed to propagate three SET commands to it over the same connection as the original handshake.

The replica receives the first SET command correctly. But the second SET command that the replica receives has lots of unnecessary zeros padded to it before and after. I tried to send hardcoded strings to the replica and that passed.

I’ve marked the task as done by mistake that’s why I can’t seem to repro logs from it again but here are the logs from the next task #HD5 which also test the same thing.

Here are the relevant logs:

[tester::#HD5] Running tests for Stage #HD5 (Replication - Multi-replica propagation)
[tester::#HD5] $ ./your_program.sh --port 6379
[tester::#HD5] [setup] Creating 3 replicas:
[tester::#HD5] [setup] 1. replica@6380 (Listening port = 6380)
[tester::#HD5] [setup] 2. replica@6381 (Listening port = 6381)
[tester::#HD5] [setup] 3. replica@6382 (Listening port = 6382)
[tester::#HD5] Creating replica@6380
[tester::#HD5] [handshake] [replica@6380] $ redis-cli PING
[tester::#HD5] [handshake] [replica@6380] ✔︎ Received "PONG"
[tester::#HD5] [handshake] [replica@6380] > REPLCONF listening-port 6380
[tester::#HD5] [handshake] [replica@6380] ✔︎ Received "OK"
[tester::#HD5] [handshake] [replica@6380] > REPLCONF capa psync2
[tester::#HD5] [handshake] [replica@6380] ✔︎ Received "OK"
[tester::#HD5] [handshake] [replica@6380] > PSYNC ? -1
[tester::#HD5] [handshake] [replica@6380] ✔︎ Received "FULLRESYNC 8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb 0"
[tester::#HD5] [handshake] [replica@6380] Received RDB file
.
. (logs of successful handshake of the other two replicas)
.
[tester::#HD5] [test] [client] $ redis-cli SET foo 123
[tester::#HD5] [test] [client] ✔︎ Received "OK"
[tester::#HD5] [test] [client] > SET bar 456
[tester::#HD5] [test] [client] ✔︎ Received "OK"
[tester::#HD5] [test] [client] > SET baz 789
[tester::#HD5] [test] [client] ✔︎ Received "OK"
[tester::#HD5] [test] Testing Replica 1/3: replica@6380
[tester::#HD5] [test] [replica@6380] Expecting "SET foo 123" to be propagated
[tester::#HD5] [test] [replica@6380] Received bytes: "*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\n123\r\n"
[tester::#HD5] [test] [replica@6380] Received RESP array: ["SET", "foo", "123"]
[tester::#HD5] [test] [replica@6380] ✔︎ Received ["SET", "foo", "123"]
[tester::#HD5] [test] [replica@6380] Expecting "SET bar 456" to be propagated
[tester::#HD5] [test] [replica@6380] Received bytes: "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\...... (this has the desired bytes somewhere in between padded with zeros from both sides)
.
.
.
[tester::#HD5] Received: "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
[tester::#HD5]            ^ error
[tester::#HD5] Error: "\x00" is not a valid start of a RESP2 value (expected +, -, :, $ or *)
[tester::#HD5] Test failed
[tester::#HD5] Terminating program

And here’s a snippet of my Java code where I’m propagating input commands as it is to replicas:

private void handleClient(Client client) throws IOException, InterruptedException {
        try {
            while (true) {
                String input = new String(client.read());
                //String output = processInput(input, client);
                if (input != null) {
                    String output = commandHandler.processInput(input, client);
                    if (output != null && !output.isBlank()) {
                        client.send(output);
                        CompletableFuture.runAsync(() -> {
                            try {
                                propagateToReplicas(input);
                            } catch (IOException e) {
                                throw new RuntimeException(e);
                            }
                        });
                    }
                }
            }
        } catch (IOException e) {
            System.out.println("IOException: " + e.getMessage());
        } catch (InterruptedException e) {
            System.out.println("InterrupedtedException");
            throw new RuntimeException(e);
        } catch (Exception e) {
            System.out.println("Caught exception: " + e.getMessage());
        } finally {
            --clientCount;
            client.closeSocket();
        }
    }

The propagateToReplicas() is implemented as below:

private void propagateToReplicas(String input) throws IOException {
    if (!RespResponseUtility.shouldSendToReplica(input)) return;

    for (int i = 0; i < replicas.size(); ++i) {
        replicas.get(i).send(input);
    }
}

The RespResponseUtility.shouldSendToReplica(input) call returns true for SET command.

replicas is a list of Clients that I’m creating of all replicas that successfully complete handshake.

private List<Client> replicas;
.
.
replicas = new ArrayList<>(); (initialised in constructor)

Here’s is where new Replicas get added to the replicas list…

void registerReplica(Client replica) {
    replicas.add(replica);
}

The above method is called on successful psync…

private String psync(String[] args, Client client) throws IOException {
        String output = RespResponseUtility.getSimpleString("FULLRESYNC " + server.getReplicationId() + " 0");
        byte[] rdbBytes = hexStringToByteArray(RDB);
        output += "$" + rdbBytes.length + "\r\n";
        client.getOutputStream().write(output.getBytes());
        client.getOutputStream().write(rdbBytes);
        //TODO: Return output from this method instead of sending respose to client directly
        server.registerReplica(client);

        return null;
    }

At last, Client class has it’s own socket and method to send data over it….

public class Client {
.
.
    private Socket socket;
.
.
    public boolean send(String output) throws IOException {
        outputStream.write(output.getBytes());
        outputStream.flush();
        return true;
    }
.
.
}

[andy1li]

Hey @lakshya-k, could you upload your code to GitHub and share the link? It will be much easier to debug if I can run it directly.

[lakshya-k]

Hi Andy,
Please find my code here: GitHub - lakshya-k/codecrafters-redis-java: codecrafters-redis-java

[lakshya-k]

Removing null characters from the input before sending it to replicas resolved the issue.

private void propagateToReplicas(String input) throws IOException {
    input = input.replaceAll("\0+$", "");
    if (!RespResponseUtility.shouldSendToReplica(input)) return;

    for (int i = 0; i < replicas.size(); ++i) {
        replicas.get(i).send(input);
    }
}

[andy1li]

@lakshya-k I added a few logs to help with debugging:

It doesn’t quite make sense that every input (even empty ones) has a length of exactly 1024. Or rather, empty message shouldn’t have been read at all.

You might want to revisit Client.read() and ensure it blocks until a message is actually received, rather than returning empty buffers of length 1024.

[lakshya-k]

I have updated Client.read()to first read the input size and then read only so much from InputStream.

public String read() throws IOException {
        InputStream inputStream = socket.getInputStream();
        int data = inputStream.read(); // * or $

        if(data == -1) return null;

        // Get input size
        int inputSize = Integer.parseInt(readNext(inputStream).strip());

        StringBuilder command = new StringBuilder("*" + inputSize + "\r\n");

        for (int i = 0; i < 2 * inputSize; ++i) {
            String word = readNext(inputStream);
            command.append(word);
        }
        return command.toString();
    }

public String readNext(InputStream inputStream) throws IOException {
        StringBuilder result = new StringBuilder();
        int data;
        while((data = inputStream.read()) != -1) {
            char c = (char) data;
            result.append((char)data);

            if (c == '\n') break;
        }
        return result.toString();
    }

Thanks for the suggestion. The issue is resolved now.