Kafka | Verify Server Connection

kafkaTemplate.send() 메소드에는 훅이 없다. 혹여나 Kafka 서버가 다운될 경우, kafkaTemplate.send() 메소드를 호출 자체를 막는게 좋은데, 마땅한 조건이 없을 경우 다음과 같은 함수를 만들어 검증해보자.

 

public boolean verifyConnection() throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrap);
        props.put("request.timeout.ms", 3000);
        props.put("connections.max.idle.ms", 5000);

        AdminClient client = AdminClient.create(props);
        Collection<Node> nodes = client.describeCluster()
            .nodes()
            .get();
        return nodes != null && nodes.size() > 0;
    }

 

또한, try catch절로 예외처리를 할 수도 있다.

    public boolean verifyConnection(String message, MessageType type) {
        
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrap);
        props.put("request.timeout.ms", 500);
        props.put("connections.max.idle.ms", 900);

        AdminClient client = AdminClient.create(props);
        try {
            client.describeCluster()
                .nodes()
                .get();
        } catch (InterruptedException | ExecutionException e) {
            
            return false;
        }
        return true;
    }

 

이 방법이 마음에 안들면 정석은 아니지만 Sockek 객체로 해당 포트가 살아있는지 확인하는 방법도 있다.

Socket socket = new Socket();
        String[] bootstrapAddress = bootstrapServers.split(":");
        try {
            socket.connect(new InetSocketAddress(bootstrapAddress[0],
                Integer.parseInt(bootstrapAddress[1])));
            socket.close();
            return true;
        } catch (Exception e) {
            log.error("[Kafka Server is not available] message = {}", e.getMessage());
            return false;
        }