To simply testing with Zookeeper on a remote Kafka cluster, one must connect to the client application ports on the backend.  When the remote Kafka cluster has multiple nodes and behind a firewall and a SSH jump server, the complexity is fairly high.  Note, the SSH jump server is the permitted man in the middle.    The client must allow application access to Zookeeper on Kafka – listening locally. Current techniques allow for a single port hosted on the developers machine for instance, 2181 listening on the local machine, and a single remote server.  This approach is not reliable – servers are taken out of service, added back, fail, or reroute to the master (another separate server).   

A typical connection looks like: 

ssh -J jump-server kafka-1 -L 2181:kafka-1:2181 "while true; do echo "waiting"; sleep 180; done"

  I worked to develop a small proxy. Setup hosts file. 1 – Edit /etc/hosts 2 – Add entry to hosts file

127.0.0.1 kafka-1
127.0.0.2 kafka-2
127.0.0.3 kafka-3
127.0.0.4 kafka-4
127.0.0.5 kafka-5

3 – Save the hosts file 4 – Setup Available interfaces (1 for each unique service) 1 is already up and in use (you only need to add the extras)

sudo ifconfig lo0 alias 127.0.0.2 up
sudo ifconfig lo0 alias 127.0.0.3 up
sudo ifconfig lo0 alias 127.0.0.4 up
sudo ifconfig lo0 alias 127.0.0.5 up

5 – Setup the port forwarding, forward to jump server ssh -L 30991:localhost:30991 jump-server 6 – Forward to Kafka server ssh -L 30991:localhost:2181 kafka-1 7 – Loop while on kafka server while true; do echo “waiting”; sleep 180; done 8 – Repeat for each kafka server increasing the port by 1 (refer to ports section for mapping) 9 – Setup the Terminal – node krb5-tcp.js 10 – Setup the Terminal – node proxy_socket.js

echo stats | nc kafka-1 2181
Zookeeper version: 3.4.6-IBM_4–1, built on 06/17/2016 01:58 GMT
Clients:
/192.168.12.47:50404[1](queued=0,recved=1340009,sent=1360508)
/192.168.12.46:48694[1](queued=0,recved=1348346,sent=1368936)
/192.168.12.48:39842[1](queued=0,recved=1341655,sent=1362178)
/0:0:0:0:0:0:0:1:39644[0](queued=0,recved=1,sent=0)

Latency min/avg/max: 0/0/2205
Received: 4878752
Sent: 4944171
Connections: 4
Outstanding: 0
Zxid: 0x1830001944e
Mode: follower
Node count: 442

11 – Use your code to access Zookeeper ServerReferences https://github.com/nodejitsu/node-http-proxy

sudo ifconfig lo0 alias 127.0.0.6 up
sudo ifconfig lo0 alias 127.0.0.7 up
sudo ifconfig lo0 alias 127.0.0.8 up

Configuration

{

"2181": {

"type": "socket",

"members": [

{ "hostname": "kafka-1", "port": 30991 },

{ "hostname": "kafka-2", "port": 30992 },

{ "hostname": "kafka-3", "port": 30993 },

{ "hostname": "kafka-4", "port": 30994 },

{ "hostname": "kafka-5", "port": 30995 }

]

}

}

Jaas Configuration
./kerberos/src/main/java/demo/kerberos/jaas.conf
TestClient {
com.sun.security.auth.module.Krb5LoginModule required
principal="ctest4@test.COM"
debug=true
useKeyTab=true
storeKey=true
doNotPrompt=false
keyTab="/Users/paulbastide/tmp/kerberos/test.headless.keytab"
useTicketCache=false;
};

Java – App.java

package demo.kerberos;

import javax.security.auth.*;
import javax.security.auth.login.*;
import javax.security.auth.callback.*;
import javax.security.auth.kerberos.*;
import java.io.*;

public class App {
    public static void main(String[] args) {

        System.setProperty("java.security.auth.login.config",
                "/Users/paulbastide/tmp/kerberos/src/main/java/demo/kerberos/jaas.conf");
        System.setProperty("java.security.krb5.conf", "/Users/paulbastide/tmp/kerberos/krb5.conf");

        Subject mysubject = new Subject();
        LoginContext lc;

        try {

            lc = new LoginContext("TestClient", mysubject, new MyCallBackHandler());
            lc.login();

        } catch (LoginException e) {
            e.printStackTrace();
        }

    }

}

Java - MyCallBackHandler.java

package demo.kerberos;

import javax.security.auth.*;
import javax.security.auth.login.*;
import javax.security.auth.callback.*;
import javax.security.auth.kerberos.*;
import java.io.*;

public class MyCallBackHandler implements CallbackHandler {
    public void handle(Callback[] callbacks)
       throws IOException, UnsupportedCallbackException {

       for (int i = 0; i < callbacks.length; i++) {
          System.out.println(callbacks[i]);
       }
    }
}

In my last few projects, I have used Test-NG.  Uniquely in my current project, I had to generate tests programmatically.  Instead of writing one test for each element in the project, I am able to generate a bunch at-will using the following pattern:

Factory

package test;

import org.testng.annotations.Factory;

public class DynamicTestFactory {


    @Factory
    public Object[] createInstances() {
        
     Object[] result = new Object[10]; 
     for (int i = 0; i < 10; i++) {
        result[i] = new ExampleProcessorTest(Integer.toString(i * 10)+ "A",Integer.toString(i*10) + "B");
      }
      return result;
    }
    
}

Test

package test;

import static org.testng.Assert.assertTrue;

import org.testng.annotations.Test;
 
public class ExampleProcessorTest {

    private String a; 
    private String b; 
    
    public ExampleProcessorTest(String a, String b) {
        this.a = a;
        this.b = b;
        
    }
    
    @Test
    public void testServer() {
        System.out.println("TEST");
        assertTrue(true);;
    }
}

Running Code

[RemoteTestNG] detected TestNG version 6.9.10
[TestNG] Running:
/private/var/folders/07/sw3n5r3170q202d5j4tx8fhw0000gn/T/testng-eclipse--1754065412/testng-customsuite.xml

TEST
TEST
TEST
TEST
TEST
TEST
TEST
TEST
TEST
TEST
PASSED: testServer
PASSED: testServer
PASSED: testServer
PASSED: testServer
PASSED: testServer
PASSED: testServer
PASSED: testServer
PASSED: testServer
PASSED: testServer
PASSED: testServer

===============================================
Default test
Tests run: 10, Failures: 0, Skips: 0
===============================================


===============================================
Default suite
Total tests run: 10, Failures: 0, Skips: 0
===============================================

[TestNG] Time taken by org.testng.reporters.EmailableReporter2@76a4d6c: 9 ms
[TestNG] Time taken by [FailedReporter passed=0 failed=0 skipped=0]: 5 ms
[TestNG] Time taken by org.testng.reporters.JUnitReportReporter@32cf48b7: 4 ms
[TestNG] Time taken by org.testng.reporters.jq.Main@130f889: 23 ms
[TestNG] Time taken by org.testng.reporters.XMLReporter@6e2c9341: 8 ms
[TestNG] Time taken by org.testng.reporters.SuiteHTMLReporter@58a90037: 10 ms

I can also trigger using testng.xml
     <class name="DynamicTestFactory" />

Reference

• https://testng.org/doc/documentation-main.html