Kafkajs: Support for Multiple A Records

[snirad32]

Hi,

I'm trying to understand if this client support one record of a broker that contains multiple A records.

Will it work in the same way as it would If I added all the brokers Ip to the brokers array in initialization?

consider: kafka-broker:9092 contain A records for kafka-broker 1-3

Will:
const kafka = new Kafka({
clientId: 'my-app',
brokers: ['kafka-broker:9092']
})
===
const kafka = new Kafka({
clientId: 'my-app',
brokers: ['kafka-broker-1:9092','kafka-broker-2:9092','kafka-broker-3:9092']
})

Thanks!

[tulios]

@snirad32 it works just fine, these brokers are only used to discover the cluster. We used to use a load balancer instead of providing a list of brokers.

[snirad32]

@tulios, I noticed that this feature is somewhat new this is why i asked:

I mean librdkafka start support it from verision 2.1:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-302+-+Enable+Kafka+clients+to+use+all+DNS+resolved+IP+addresses

can you please point out how you do it in your library, I tried to find it but I couldnt... thanks!

[tulios]

We discussed this, and what I wrote before is not true. But that's a small change on our side. I am wrapping up another issue now, but I can jump into this one once I'm done since it's a small change.

[snirad32]

you can point me to the place that needs fixing and I can do it myself if you want :)

[tulios]

Hi @snirad32, you might need a new property to enable this behavior, but I think the change should happen in connection#connect - https://github.com/tulios/kafkajs/blob/master/src/network/connection.js#L97

Another option is to add the information about multiple IPs to the connection and handle the error in the broker#connect - https://github.com/tulios/kafkajs/blob/master/src/broker/index.js#L82

[unbalanced]

Hey @tulios i believe that it's a very important feature for anyone who's aiming to work with KafkaJS on a large kafka cluster. Could you shed light on if/when do you expect it to be added?

Many Thanks.

[Nevon]

Unfortunately we all have jobs and responsibilities outside of KafkaJS, so it's hard to say when exactly any of us will be able to work on an issue that isn't a pressing need for us.

Ultimately, the fastest way to get a certain feature in is to make a pull request implementing it. It's generally easier for us to to make the time to review PRs than create them ourselves, and it creates more of an urgency since we want to be receptive and appreciative of other people's time and contributions.

[tulios]

Hey @unbalanced, I will take a look at it. It shouldn't be too hard to support it.

[t-d-d]

Is this the
*client.dns.lookup == *use_all_dns_ips
option from the java client?

On Mon, 10 Feb 2020, 17:25 Túlio Ornelas, [e-mail] wrote:

Hey @unbalanced https://github.com/unbalanced, I will take a look at
it. It shouldn't be too hard to support it.

—
You are receiving this because you are subscribed to this thread.
Reply to this email directly, view it on GitHub
https://github.com/tulios/kafkajs/issues/509?email_source=notifications&email_token=ABDLW5Q7XL35OXD6POWJ5CDRCEMPBA5CNFSM4I2XSUXKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOELHZM6Y#issuecomment-584029819,
or unsubscribe
https://github.com/notifications/unsubscribe-auth/ABDLW5R3ZFNIQ4KQOTZS4IDRCEMPBANCNFSM4I2XSUXA
.

[unbalanced]

Thanks everyone,
@tulios, this is much appreciated.

@Nevon, totally understand! I wanted to see first what's up before I dive in further. Since from a glimpse on the code on master it looked to me as if it should already work. And a simulation I made showed an inconclusive result.

[Nevon]

Is this the
client.dns.lookup == use_all_dns_ips
option from the java client?

Yes, correct, that would be the equivalent change. For reference, that was implemented in this PR
for the Java client: https://github.com/apache/kafka/pull/4987

I took a brief look at this this morning, to see if it was a trivial change or not. My assessment is that while it's not _huge_, it's not _that_ simple. Note that I'm not saying that I'm going to implement this - I'm just sharing what I learned looking through this.

Here's how things currently work:

1. Connection.connect calls createSocket, passing in on* (onConnect, onError etc.) and is also responsible for enforcing the connection timeout. onError will disconnect, clear queued up requests, abort ongoing authentications and a few other things. When onConnect is called, Connection.connect will resolve.
2. createSocket simply calls the provided SocketFactory and expects to get a net.Socket back, to which it attaches the on* handlers.
3. SocketFactory can be passed in from userland to allow people to customize the socket, so we can't really modify what's going on in there, but we know it will receive a host and a port and return a net.Socket instance.
4. The default SocketFactory that we ship calls net.connect, which is an alias of net.createConnection. What's notable is that if it is called with a hostname rather than an IP address, it will call dns.lookup with hardcoded options that make it resolve to a single IP.

What this means is that we cannot rely on this being handled within net.Socket itself, and since SocketFactory can be injected we cannot solve it in there, which means that it has to be solved either in createSocket or Connection.connect. I.e. we have to resolve the hosts ourselves.

Another interesting thing that we haven't mentioned is dealing with IPs resolving to the same underlying host. For example, if you currently specify localhost as a broker, it will most likely resolve to 127.0.0.1, and we will make a single connection attempt to that IP. If we naively change to using all resolved IP addresses this behavior will change so that we make two connection attempts. Why? Because of multiple IP families:

// Current behavior
dns.lookup('localhost', (err, addr) => console.log(addr))
> 127.0.0.1

// After change
dns.lookup('localhost', { all: true }, (err, addr) => console.log(addr))
> [ { address: '127.0.0.1', family: 4 }, { address: '::1', family: 6 } ]

The Java client deals with this by taking the first resolved IP family and then filtering out all other resolved IPs that are not of the same family. If you combine this with the verbatim: true option on lookup, we will basically let the DNS server decide if we should use IPv4 or IPv6. I'm not sure what the implications of this are, but it does start to feel like maybe this wouldn't be a completely transparent change.

Another consideration is that using the custom SocketFactory we currently allow users to replace dns.lookup with for example dns.resolve. This can have important consequences, and so I think it's an important capability to retain now that we are lifting DNS resolution up one level. Therefore I'm inclined to say that this should be a top-level option on the client.

new Kafka({
  dns?: {
    all?: boolean, // default false
    lookup?: (hostname: string, options: LookupOptions, callback: (err: Error | null, address: LookupAddress[], family: number) => void): void // default `dns.lookup`
  }
})

---

My general thinking is that this probably has to be solved within Connection.connect, but that code is currently a little bit hairy, so it's not as straightforward as I would like. Basically, before connecting we'd need to lookup the addresses for the host. If dns.all: false we take addresses.slice(0,1), otherwise the whole lot. We then filter out any addresses other than the one of the first family returned.

On connect, we essentially loop through the addresses and try to connect to each one. Only when we have exhausted all of them do we propagate the error upwards. This means re-writing the error handling as well as the timeout enforcement.