kozlovic
commented
1 month ago @roeschter Are you sure of the setup? I tried to write a test based on your description and it passes. I just check that each queue sub receives at least 10% of the messages, but we usually test for close to even distribution. But since you claim that one does not receive any message, the 10% should be enough. Here is the test, let me know what changes to make to reproduce the issue:
func TestLeafNodeQueueGroupDistribution(t *testing.T) {
hc := createClusterWithName(t, "HUB", 3)
defer hc.shutdown()
// Now have a cluster of leafnodes with each one connecting to corresponding HUB(n) node.
c1 := `
server_name: LEAF1
listen: 127.0.0.1:-1
cluster { name: ln22, listen: 127.0.0.1:-1 }
leafnodes { remotes = [{ url: nats-leaf://127.0.0.1:%d }] }
`
lconf1 := createConfFile(t, []byte(fmt.Sprintf(c1, hc.opts[0].LeafNode.Port)))
ln1, lopts1 := RunServerWithConfig(lconf1)
defer ln1.Shutdown()
c2 := `
server_name: LEAF2
listen: 127.0.0.1:-1
cluster { name: ln22, listen: 127.0.0.1:-1, routes = [ nats-route://127.0.0.1:%d] }
leafnodes { remotes = [{ url: nats-leaf://127.0.0.1:%d }] }
`
lconf2 := createConfFile(t, []byte(fmt.Sprintf(c2, lopts1.Cluster.Port, hc.opts[1].LeafNode.Port)))
ln2, _ := RunServerWithConfig(lconf2)
defer ln2.Shutdown()
c3 := `
server_name: LEAF3
listen: 127.0.0.1:-1
cluster { name: ln22, listen: 127.0.0.1:-1, routes = [ nats-route://127.0.0.1:%d] }
leafnodes { remotes = [{ url: nats-leaf://127.0.0.1:%d }] }
`
lconf3 := createConfFile(t, []byte(fmt.Sprintf(c3, lopts1.Cluster.Port, hc.opts[2].LeafNode.Port)))
ln3, _ := RunServerWithConfig(lconf3)
defer ln3.Shutdown()
// Check leaf cluster is formed and all connected to the HUB.
lnServers := []*Server{ln1, ln2, ln3}
checkClusterFormed(t, lnServers...)
for _, s := range lnServers {
checkLeafNodeConnected(t, s)
}
// Check each node in the hub has 1 connection from the leaf cluster.
for i := 0; i < 3; i++ {
checkLeafNodeConnectedCount(t, hc.servers[i], 1)
}
// Create a client and qsub on LEAF1 and LEAF2.
nc1 := natsConnect(t, ln1.ClientURL())
defer nc1.Close()
var qsub1Count atomic.Int32
natsQueueSub(t, nc1, "foo", "queue1", func(_ *nats.Msg) {
qsub1Count.Add(1)
})
nc2 := natsConnect(t, ln2.ClientURL())
defer nc2.Close()
var qsub2Count atomic.Int32
natsQueueSub(t, nc1, "foo", "queue1", func(_ *nats.Msg) {
qsub2Count.Add(1)
})
sendAndCheck := func(idx int) {
t.Helper()
nchub := natsConnect(t, hc.servers[idx].ClientURL())
defer nchub.Close()
total := 1000
for i := 0; i < total; i++ {
natsPub(t, nchub, "foo", []byte("from hub"))
}
checkFor(t, time.Second, 15*time.Millisecond, func() error {
if trecv := int(qsub1Count.Load() + qsub2Count.Load()); trecv != total {
return fmt.Errorf("Expected %v messages, got %v", total, trecv)
}
return nil
})
// Now that we have made sure that all messages were received,
// check that qsub1 and qsub2 are getting at least some.
if n := int(qsub1Count.Load()); n <= total/10 {
t.Fatalf("Expected qsub1 to get some messages, but got %v", n)
}
if n := int(qsub2Count.Load()); n <= total/10 {
t.Fatalf("Expected qsub2 to get some messages, but got %v", n)
}
// Reset the counters.
qsub1Count.Store(0)
qsub2Count.Store(0)
}
// Send from HUB1, HUB2 and HUB3
for i := 0; i < 3; i++ {
sendAndCheck(i)
}
}
derekcollison
roeschter
Observed behavior
From two subscribers in a queue group connected to two different nodes in a leaf cluster only one will receive messages IFF those messages arrive from a hub cluster to which the leaf cluster is connected. The second subscribers (initially idle) will receive messages once the primary is terminated. This is reversible (restart primary connected to the original node and will take over again).
Expected behavior
All subcribers in the same queue group a leaf cluster are treated equally.
Server and client version
Server 2.10.21 Nats - compiled from main
Host environment
Windows/Mac/Linux - reproduced by customer, Borja and me
Steps to reproduce
To reproduce deterministically, all leaf node connections and nats-cli connections will connect to specific nodes.
nats --server LEAF1 sub --queue=q1 foonats --server LEAF2 sub --queue=q1 foonats --server HUB1 pub foo Hello//Only the subscriber on LEAF2 will receive messagesnats --server HUB3 pub foo Hello//Both subscribers receive messagesTentative explanation: The "prefer local cluster listeners in work queues" logic is not leaf node aware. When the message is published such that both listeners have the same "distance" (routing hops) to the publisher the work queue LB works. When the message is published such that both listeners have the different "distance" to the publisher the work queue LB fails.