Option 1:

• In KafkaComputerLauncher invoke SlaveComputer#setChannel()
  • Current interface: InputStream in, OutputStream out, TaskListener taskListener, Channel.Listener listener
  • We need need to produce KafkaRemotingOutputStream(KafkaProducer) and KafkaRemotingInputStream(KafkaConsumer)
  • It is not efficient, but should be a good start

Option 2:

• Patches in Jenkins core may be needed (unhardcode Channel Builder)
• CommandTransport for Remoting Kafka. Get Kafka Consumer and Producer connected into it
  • It is possible to connect new command transport to the existing Kafka instances
  • https://github.com/jenkinsci/remoting/blob/master/src/main/java/hudson/remoting/CommandTransport.java
• RemotingKafkaChannelBuilder
  • https://github.com/jenkinsci/remoting/blob/master/src/main/java/hudson/remoting/Channel.java
  • Constructor: @Nonnull ChannelBuilder settings, @Nonnull CommandTransport transport
  • We pass KafkaCommandTransport
  • The default ChannelBuilder should be fine
  • Usage example: SlaveComputer#setChannel()
• In KafkaComputerLauncher invoke SlaveComputer#setChannel()
  • Current interface: InputStream in, OutputStream out, TaskListener taskListener, Channel.Listener listener
  • We need need to produce KafkaRemotingOutputStream(KafkaProducer) and KafkaRemotingInputStream(KafkaConsumer)
  • It is not efficient, but should be a good start

Core patch would be needed here:

• Support passing ChannelBuilder

public void setChannel(InputStream in, OutputStream out, OutputStream launchLog, Channel.Listener listener) throws IOException, InterruptedException {
        ChannelBuilder cb = new ChannelBuilder(nodeName,threadPoolForRemoting)
            .withMode(Channel.Mode.NEGOTIATE)
            .withHeaderStream(launchLog);

        for (ChannelConfigurator cc : ChannelConfigurator.all()) {
            cc.onChannelBuilding(cb,this);
        }

        Channel channel = cb.build(in,out);
        setChannel(channel,launchLog,listener);
    }

Option 3. Create KafkaComputer and maybe copy-paste some code?

• Close to option 2, but without patching the core
• KafkaCommandTransport as in Phase 2
• KafkaChannelBuilder as in Option 2, which creates channel with KafkaCommandTransport
• Creates a new Computer type (Kafka Agent), which is fine
  • "KafkaPermanentAgent" as a name?
  • It can inherit SlaveComputer
  • We need to replace setChannel() implementation by one using KafkaCommandTransport (see below)
    • It overrides old stream-based implementations and throw IOException there
• Oleg: We need to check that TcpAgentListener won't dare to invoke the agent if somebody tries to connect. Oleg to check it. It is for the next phase, no need to do it now
• Connection: KafkaComputerLauncher invokes the new setChannel() implementation

public void setChannel(OutputStream launchLog, Channel.Listener listener) throws IOException, InterruptedException {
        ChannelBuilder cb = new KafkaChannelBuilder(nodeName, threadPoolForRemoting)
            .withMode(Channel.Mode.NEGOTIATE)
            .withHeaderStream(launchLog);

        for (ChannelConfigurator cc : ChannelConfigurator.all()) {
            cc.onChannelBuilding(cb,this);
        }

        Channel channel = cb.build(in,out);
        setChannel(channel, launchLog, listener);
    }