Write deduplicated 9-digit numbers from concurrent clients to a log

Question

Problem Statement:

Write a server (“Application”) in Java that opens a socket and restricts input to at most 5 concurrent clients. Clients will connect to the Application and write any number of 9 digit numbers, and then close the connection. The Application must write a de-duplicated list of these numbers to a log file in no particular order.

Primary Considerations

• The Application should work correctly as defined below in Requirements.
• The overall structure of the Application should be simple.
• The code of the Application should be descriptive and easy to read, and the build method and runtime parameters must be well-described and work.
• The design should be resilient with regard to data loss.
• The Application should be optimized for maximum throughput, weighed along with the other Primary Considerations and Requirements below.

Requirements

1. The Application must accept input from at most 5 concurrent clients on TCP/IP port 4000.
2. Input lines presented to the Application via its socket must either be composed of exactly nine decimal digits (e.g.: 314159265 or 007007009) immediately followed by a server-native newline sequence; or a termination sequence as detailed in #9, below.
3. Numbers presented to the Application must include leading zeros as necessary to ensure they are each 9 decimal digits.
4. The log file, to be named "numbers.log”, must be created anew and/or cleared when the Application starts.
5. Only numbers may be written to the log file. Each number must be followed by a server-native newline sequence.
6. No duplicate numbers may be written to the log file.
7. Any data that does not conform to a valid line of input should be discarded and the client connection terminated immediately and without comment.
8. Every 10 seconds, the Application must print a report to standard output:
   • The difference since the last report of the count of new unique numbers that have been received.
   • The difference since the last report of the count of new duplicate numbers that have been received.
   • The total number of unique numbers received for this run of the Application.
   • Example text for #8: Received 50 unique numbers, 2 duplicates. Unique total: 567231
9. If any connected client writes a single line with only the word "terminate" followed by a server-native newline sequence, the Application must disconnect all clients and perform a clean shutdown as quickly as possible.
10. Clearly state all of the assumptions you made in completing the Application along with any instructions on how to set up and run it in a README file.

Notes

• You may write tests at your own discretion. Tests are useful to ensure your Application passes Primary Consideration A.
• You may use common libraries in your project such as Apache Commons and Google Guava, particularly if their use helps improve Application simplicity and readability. However the use of large frameworks, such as Akka, is prohibited.
• Your Application may not for any part of its operation use or require the use of external systems, for example Apache Kafka or Redis.
• At your discretion, leading zeroes present in the input may be stripped—or not used—when writing output to the log or console.
• Robust implementations of the Application typically handle more than 2M numbers per 10-second reporting period on a modern MacBook Pro laptop (e.g.: 16 GiB of RAM and a 2.5 GHz Intel i7 processor).

My implementation:

Application.java:

public class Application {
    static final int PORT = 4000;

    public static void main(String[] args) {
        Server server = new Server();
        System.out.println("Starting server");
        server.start(PORT);
    }
}

Server.java:

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.regex.Pattern;

public class Server {
    private static final int MAX_CLIENTS = 5;
    private ServerSocket serverSocket;
    private ExecutorService executorService = Executors.newFixedThreadPool(MAX_CLIENTS);

    private FileWriter fw;
    private BufferedWriter bw;
    BlockingQueue blockingQueue = new LinkedBlockingQueue();


    public void start(int port) {
        try {
            serverSocket = new ServerSocket(port);
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            fw = new FileWriter("numbers.log", false);
        } catch (IOException e) {
            e.printStackTrace();
        }
        bw = new BufferedWriter(fw);
        LogWriterTask logWriterTask = new LogWriterTask(blockingQueue, bw);
        logWriterTask.start();


        while (true) {
            try {
                ClientHandler clientHandler = new ClientHandler(serverSocket.accept(), blockingQueue);
                executorService.submit(clientHandler);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private class ClientHandler extends Thread {
        private Socket clientSocket;
        private BufferedReader in;
        BlockingQueue<Integer> blockingQueue;
        private Pattern numberPattern = Pattern.compile("\\d{9}");

        public ClientHandler(Socket socket, BlockingQueue<Integer> blockingQueue) {
            this.clientSocket = socket;
            this.blockingQueue = blockingQueue;
        }

        @Override
        public void run() {

            try {
                in = new BufferedReader(
                        new InputStreamReader(clientSocket.getInputStream()));

                String inputLine = "";

                System.out.println("Client connection started");
                while (true) {

                    try {
                        inputLine = in.readLine();
                        if (inputLine == null) {
                            break;
                        }
                        if (inputLine.equals("terminate")) {    //Disconnect all clients
                            System.exit(0);
                        };
                    } catch (SocketException e) {
                        e.printStackTrace();
                        stopClient();
                        break;
                    }

                    int num = processNumber(inputLine);
                    if (num == -1) {
                        stopClient();
                        break;
                    }
                    blockingQueue.add(num);
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                stopClient();
            }
        }


        public void stopClient() {
            try {
                in.close();
                clientSocket.close();
                System.out.println("Client connection closed");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private int processNumber(String inputLine) {
            if (!numberPattern.matcher(inputLine).matches()) return -1; //Invalid format, terminate client

            int num;

            try {
                num = Integer.parseInt(inputLine);
            } catch(NumberFormatException e) {  //Invalid format, terminate client
                num = -1;
            }

            return num;
        }
    }


}

LogWriterTask.java:

import java.io.BufferedWriter;
import java.io.IOException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;

public class LogWriterTask extends Thread {
    private final static int MAX_NINE_DIGIT_RANGE = 1000000000;
    private final static int SUMMARY_WAIT_PERIOD = 10000;
    private int uniqueCount = 0;
    private int duplicateCount = 0;
    private int uniqueTotal = 0;
    private BlockingQueue<Integer> blockingQueue;
    private BufferedWriter bw;
    private int[] uniqueNums = new int[MAX_NINE_DIGIT_RANGE];

    public LogWriterTask(BlockingQueue<Integer> blockingQueue, BufferedWriter bw) {
        this.bw = bw;
        this.blockingQueue = blockingQueue;
        Timer timer = new Timer();
        timer.schedule(new SummaryTask(), 0, SUMMARY_WAIT_PERIOD);
    }

    @Override
    public void run() {
        while (true) {
            while (!blockingQueue.isEmpty()) {
                int num = blockingQueue.poll();
                if (uniqueNums[num] == 0) {
                    try {
                        uniqueCount++;
                        uniqueTotal++;
                        bw.write(String.format("%09d", num));
                        bw.newLine();
                        bw.flush();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                } else {
                    duplicateCount++;
                }

                uniqueNums[num]++;
            }
        }
    }

    class SummaryTask extends TimerTask {
        @Override
        public void run() {
            System.out.printf("Received %d unique numbers, %d duplicates. Unique total: %d\n", uniqueCount, duplicateCount, uniqueTotal);
            uniqueCount = 0;
            duplicateCount = 0;
        }
    }
}

Client.java:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

public class Client {
    private Socket clientSocket;
    private PrintWriter out;
    private BufferedReader in;

    public void startConnection(String ip, int port) {
        try {
            clientSocket = new Socket(ip, port);
            out = new PrintWriter(clientSocket.getOutputStream(), true);
            in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
        } catch (IOException e) {
            System.err.println("Error when initializing connection");
        }

    }

    public void sendMessage(String msg) {
        try {
            out.println(msg);
            return;
        } catch (Exception e) {
            return;
        }
    }

    public void stopConnection() {
        try {
            in.close();
            out.close();
            clientSocket.close();
        } catch (IOException e) {
            System.err.println("error when closing");
        }

    }

Testing the solution:

Client.java:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

public class Client {
    private Socket clientSocket;
    private PrintWriter out;
    private BufferedReader in;

    public void startConnection(String ip, int port) {
        try {
            clientSocket = new Socket(ip, port);
            out = new PrintWriter(clientSocket.getOutputStream(), true);
            in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
        } catch (IOException e) {
            System.err.println("Error when initializing connection");
        }

    }

    public void sendMessage(String msg) {
        try {
            out.println(msg);
            return; //in.readLine();
        } catch (Exception e) {
            return;
        }
    }

    public void stopConnection() {
        try {
            in.close();
            out.close();
            clientSocket.close();
        } catch (IOException e) {
            System.err.println("error when closing");
        }

    }

ApplicationTest.java:

import org.junit.jupiter.api.Test;

public class ApplicationTest {
    private static final String LOCAL_HOST = "127.0.0.1";
    private static final int PORT = 4000;

    @Test
    public void testThroughput() {
        Client client = new Client();
        client.startConnection(LOCAL_HOST, PORT);

        Client client2 = new Client();
        client2.startConnection(LOCAL_HOST, PORT);

        for (int i = 0; i < 50000; i++) {
            int minNum = 0;
            int maxNum = 999999999;
            int random = (int) ((Math.random() * (maxNum - minNum)) + minNum);
            String randomStr = String.format("%09d", random);
            client.sendMessage(randomStr);

            int random2 = (int) ((Math.random() * (maxNum - minNum)) + minNum);
            String randomStr2 = String.format("%09d", random2);
            client.sendMessage(randomStr2);
        }
    }

    @Test
    public void testTerminateSequence() {
        Client client = new Client();
        client.startConnection(LOCAL_HOST, 4000);
        client.sendMessage("123456789");
        client.sendMessage("terminate");
    }

    @Test
    public void testInvalidInput() {
        Client client = new Client();
        client.startConnection(LOCAL_HOST, 4000);
        client.sendMessage("123456789");
        client.sendMessage("123");
        client.sendMessage("123456789");
    }

    @Test
    public void testDuplicates() {
        Client client = new Client();
        client.startConnection(LOCAL_HOST, 4000);
        client.sendMessage("123456789");
        client.sendMessage("987654321");
        client.sendMessage("123456789");
        client.sendMessage("987654321");
    }

}

Answer 1

I'm reasonable sure you've submitted it by now.

General

Members should be as private as possible. In application, PORT should be private scope, not default.

Classes not designed for extension should be marked final. Member variables which will not be reassigned should be marked final.

Classes not designed to be constructed (Application) should have a private constructor to prevent it.

Java allows the use of _ as a separator in numbers. 10_000 is easier to read than 10000.

The typical ordering for variable modifiers would be private static final, not private final static. Worse, it's inconsistent between classes.

Variable name abbreviations should be avoided. For instance, bw is meaningless.

Extending Thread is almost always wrong. You're not making a new, generally useful thread type. You're just specifying what the thread should run. Prefer implements Runnable.

BlockingQueue should always specify its generic type. In Server it does not.

Prefer try-with-resources to try/catch.

Minimize scope as much as possible. You sometimes declare variables with greater scope than they need.

This is stylistic, but it's generally frowned upon in java to omit curly braces and put a return on the same line.

Consistent use of whitespace is important. It shows attention to detail.

Server

The FileWriter and BufferedWriter variables are only used in one method. They don't even belong in this file. They should be in LogWriterTask.

You never close the file you're opening on the server. That will wind up as a resource leak when a client sends terminate and you call System.exit(). It definitely does not meet requirement #9. You'll need to pass the ExecutorService into the ClientHandler, call shutdownNow(), and then have your ClientHandler correctly interrupt.

Minimize the code inside try-catch blocks. The inputLine checks don't throw SocketException and can be outside it.

You have an extra ; after the terminate if block.

stopClient would be better named closeSocket

processNumber's num variable is not useful.

String matching is inefficient. Check the length, then convert it to an integer, then make sure it's positive.

If you don't expect the data to contain invalid inputs, I'd say an exception is more warranted than a magic return value. If it's expected, than the -1 is more reasonable.

Pattern is thread-safe and can be made static.

stopClient should be private. It is not accessed outside the class that contains it.

The inner class ClientHandler should be static, because it does not rely on the scope of the class that contains it. It's probably big enough to be its own class.

LogWriterTask It would be preferable to scope the file writer to this class. The `Server` doesn't need to know or care about it.

Don't loop on true, checking if it's empty, and then poll. That's a busy-wait loop, where CPU is being burned spinning through the loop doing nothing. Instead use the blocking method take() in the while loop.

Using a boolean[] would be better. Using some sort of bit field, such as a BitSet, would be much better. You don't need to count the total number of duplicates, just whether something has been duplicated. BitSet's cardinality() method will track total uniques for you.

timer.schedule() will not run every ten seconds. It will run every ten seconds from the completion of the previous run. Perhaps you want scheduleAtFixedRate?

The problem with using uniqueCount, duplicateCount, and uniqueTotal is that you can get off-by-one errors if the SummaryTask runs while the LogWriterTask is in the middle of running. For instance, if LogWriterTask yields between uniqueCount and uniqueTotal getting set, the total will be one lower than it should. You could also have a situation where the values have been updated, but the number has not yet been written to the file. You can fix that by synchronizing or using a locking object, but it's a performance vs. correctness tradeoff. You at least need to synchronize the reads and writes to the variables that track state.

The LogWriterTask will not halt when terminate is called. You can address this by adding it to the ExecutorService, so that when that terminates it gets interrupted also.

If you made all these changes, your code might look something like:

LogWriterTask:

public final class LogWriterTask implements Runnable {

    private static final int SUMMARY_WAIT_PERIOD = 10_000;
    private final BlockingQueue<Integer> blockingQueue;
    private final BitSet bitSet = new BitSet(1_000_000_000);
    private final Object lock = new Object();
    private int uniqueCount = 0;
    private int duplicateCount = 0;
    private int uniqueTotal = 0;

    public LogWriterTask(BlockingQueue<Integer> blockingQueue) {
        this.blockingQueue = blockingQueue;
        Timer timer = new Timer();
        timer.scheduleAtFixedRate(new SummaryTask(), 0, SUMMARY_WAIT_PERIOD);
    }

    @Override
    public void run() {

        try (FileWriter fileWriter = new FileWriter("numbers.log", true);
                BufferedWriter outputWriter = new BufferedWriter(fileWriter)) {

            while (true) {
                int number = blockingQueue.take();
                synchronized(lock) {
                    if (bitSet.get(number)) {
                        duplicateCount++;
                        continue;
                    }
                    bitSet.set(number);
                    uniqueCount++;
                    uniqueTotal++;
                    try {
                        outputWriter.write(String.format("%09d", number));
                        outputWriter.newLine();
                        outputWriter.flush();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            System.err.println("LogWriterTask interrupted");
        }
    }

    final class SummaryTask extends TimerTask {
        @Override
        public void run() {
            synchronized(lock) {
                System.out.printf("Received %d unique numbers, %d duplicates. Unique total: %d\n",
                        uniqueCount, duplicateCount, uniqueTotal);
                uniqueCount = 0;
                duplicateCount = 0;
            }
        }
    }
}

Server:

public final class Server {

    private static final int MAX_CLIENTS = 5;
    private final BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();
    private final ExecutorService executorService = Executors.newFixedThreadPool(MAX_CLIENTS + 1);
    private ServerSocket serverSocket;

    public void start(int port) {
        try {
            serverSocket = new ServerSocket(port);
        } catch (IOException e) {
            e.printStackTrace();
        }

        LogWriterTask logWriterTask = new LogWriterTask(blockingQueue);
        executorService.submit(logWriterTask);

        while (true) {
            try {
                ClientHandler clientHandler = new ClientHandler(serverSocket.accept(), blockingQueue, executorService);
                executorService.submit(clientHandler);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private static class ClientHandler implements Runnable {

        private final BlockingQueue<Integer> blockingQueue;
        private final Socket clientSocket;
        private final ExecutorService executorService;

        public ClientHandler(Socket socket, BlockingQueue<Integer> blockingQueue, ExecutorService executorService) {
            this.clientSocket = socket;
            this.blockingQueue = blockingQueue;
            this.executorService = executorService;
        }

        @Override
        public void run() {

            try (BufferedReader in = new BufferedReader(
                    new InputStreamReader(clientSocket.getInputStream()))) {

                System.out.println("Client connection started");
                while (true) {
                    if (Thread.currentThread().isInterrupted()) {
                        break;
                    }

                    String inputLine = "";
                    try {
                        inputLine = in.readLine();
                    } catch (SocketException e) {
                        e.printStackTrace();
                        break;
                    }

                    if (inputLine == null) {
                        break;
                    }

                    if (inputLine.equals("terminate")) {
                        executorService.shutdownNow();
                        break;
                    }

                    blockingQueue.add(parseInt(inputLine));
                }
            } catch (IOException | NumberFormatException e) {
                e.printStackTrace();
            } finally {
                closeSocket();
            }
        }

        private int parseInt(String inputLine)
                throws NumberFormatException {
            if (inputLine.length() != 9) {
                throw new NumberFormatException("Input has invalid length: '" + inputLine + "'");
            }

            int number = Integer.parseInt(inputLine);

            if (number < 0) {
                throw new NumberFormatException("Input contained a minus sign: '" + inputLine + "'");
            }

            return number;
        }

        private void closeSocket() {
            try {
                clientSocket.close();
                System.out.println("Client connection closed");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

Answer 2

Thanks for sharing your code. It is a pretty good approach to the problem, so I have just a few point I'd like to mention.

"In band signalling"

Your code uses a special value from the quantity to signal a special condition. As many other OO languages, java has the Exception concept for this.

Yes, You should not use Exceptions to replace control flow but this is clearly not the case here since the "changed control flow" is not in the same method but some way up in the call stack. The fact that you added a comment explicitly mentioning the intended exceptional behavior supports my claim.

Use of break

The break instruction terminates the current loop. But in a larger method like yours, how do you (quickly) find the point were the execution continues after the break?

The better way to deal with that is to move the loop into am separate method and use return instead of break.

In your special case the break (and the if preceding it) would be completely obsolete if you signaled the the Termination of the Thread by an Exception.

Closing resources

Java has try with resource that handles the closing of most kind of resources for you.

Beside being less clutter it also resolves the Problem of "swallowed Exceptions" When you reach the finally block doing the cose() calls after an Exception has occurred and the closing fails with an Exception too, only the original exception is not propagated and cannot displayed to the user. try with resource always propagates the original Exception in that case automatically.