Check version when serverProperties is not empty and < 3.11.0 #345

Gsantomaggio · 2024-07-29T10:00:34Z

Fixes: #344

When the serverProperties["version"] is not empty the client now checks the version.
If the version is >=3.11 it is possible to enable the exchangeVersion function.

Fixes another small bug when the single active consumer is enabled

cc @hiimjako

and is less than 3.11 Signed-off-by: Gabriele Santomaggio <[email protected]>

Gsantomaggio · 2024-07-29T11:56:15Z

@LucasZhanye can you please try this PR?

LucasZhanye · 2024-07-30T02:01:06Z

@LucasZhanye can you please try this PR?

Yes, I can

I try it later, and I will reply it after I finished

LucasZhanye · 2024-07-30T02:52:24Z

@Gsantomaggio I try it, it is work well!
the output log is : 2024/07/30 10:49:39 [info] - Server version is less than 3.11.0, skipping command version exchange.

And I want to know how long it will release a new version for it ?

LucasZhanye · 2024-07-30T03:26:12Z

@LucasZhanye can you please try this PR?

oh, I find another problem.

Now I can DeclareStream,NewProducer,but producer.Send is not take effect.

my code is :

const (
	EVENTSTREAM = "events"
)

type Event struct {
	Name string
}

func main() {
	// COnnect to the Stream Plugin on Rabbimq
	env, err := stream.NewEnvironment(
		stream.NewEnvironmentOptions().
			SetHost("10.2.3.200").
			SetPort(15552).
			SetUser("guest").
			SetPassword("guest"))
	if err != nil {
		panic(err)
	}
	// Declare the stream, Set segmentsize and Maxbytes on stream
	err = env.DeclareStream(EVENTSTREAM, stream.NewStreamOptions().
		SetMaxSegmentSizeBytes(stream.ByteCapacity{}.MB(1)).
		SetMaxLengthBytes(stream.ByteCapacity{}.MB(2)))

	if err != nil {
		panic(err)
	}

	// Create a new Producer
	producerOptions := stream.NewProducerOptions()
	producerOptions.SetProducerName("producer")

	// Batch 100 Events in the same Frame, and the SDK will handle everything
	// DEDPULICATION DOES NOT WORK WITH SUBENTRY
	producerOptions.SetSubEntrySize(100)
	producerOptions.SetCompression(stream.Compression{}.Gzip())

	producer, err := env.NewProducer(EVENTSTREAM, producerOptions)
	if err != nil {
		panic(err)
	}

	// Publish 6001 messages
	for i := 0; i <= 6001; i++ {
		// fmt.Println("i = ", i)
		event := Event{
			Name: "test",
		}

		data, err := json.Marshal(event)
		if err != nil {
			panic(err)
		}

		message := amqp.NewMessage(data)
		// Apply properties to our message
		props := &amqp.MessageProperties{
			CorrelationID: uuid.NewString(),
		}
		message.Properties = props
		// Set Publishing ID to prevent Deduplication
		message.SetPublishingId(int64(i))
		// Sending the message
		if err := producer.Send(message); err != nil {
			panic(err)
		}
	}

	producer.Close()
}

But the rabbitmq management show as below:

the queue message is zero,but I send 6002

Gsantomaggio · 2024-07-30T07:14:25Z

@LucasZhanye, I cannot reproduce the issue.

You should check the server logs and also try to remove these values:

SetMaxSegmentSizeBytes(stream.ByteCapacity{}.MB(1)).
SetMaxLengthBytes(stream.ByteCapacity{}.MB(2)))

also you should always enable the confirmation and wait before close the producer since the send is async

	producerOptions := stream.NewProducerOptions()
	producerOptions.SetProducerName("producer")

	// Batch 100 Events in the same Frame, and the SDK will handle everything
	// DEDPULICATION DOES NOT WORK WITH SUBENTRY
	producerOptions.SetSubEntrySize(100)
	producerOptions.SetCompression(stream.Compression{}.Gzip())

	producer, err := env.NewProducer(EVENTSTREAM, producerOptions)
	if err != nil {
		panic(err)
	}
	chPublishConfirm := producer.NotifyPublishConfirmation()
	totalConfirmed := int32(0)
	chFinished := make(chan struct{})
	go func() {
		for confirmed := range chPublishConfirm {
			for _, msg := range confirmed {
				if msg.IsConfirmed() {
					atomic.AddInt32(&totalConfirmed, 1)
					if atomic.LoadInt32(&totalConfirmed) == 6001 {
						chFinished <- struct{}{}
					}
				} else {
					fmt.Printf("message %s failed \n  ", msg.GetMessage().GetData())
				}

			}
		}
	}()

	// Publish 6001 messages
	for i := 0; i <= 6001; i++ {
		// fmt.Println("i = ", i)
		event := Event{
			Name: "test",
		}

		data, err := json.Marshal(event)
		if err != nil {
			panic(err)
		}

		message := amqp.NewMessage(data)
		// Apply properties to our message
		props := &amqp.MessageProperties{
			CorrelationID: uuid.NewString(),
		}
		message.Properties = props
		// Set Publishing ID to prevent Deduplication
		message.SetPublishingId(int64(i))
		// Sending the message
		if err := producer.Send(message); err != nil {
			panic(err)
		}
	}

	<-chFinished

	producer.Close()
	env.Close()
	fmt.Printf("Closed with total Confirmed: %d\n", totalConfirmed)

LucasZhanye · 2024-07-30T07:32:05Z

@LucasZhanye, I cannot reproduce the issue.

You should check the server logs and also try to remove these values:

SetMaxSegmentSizeBytes(stream.ByteCapacity{}.MB(1)).
SetMaxLengthBytes(stream.ByteCapacity{}.MB(2)))

also you should always enable the confirmation and wait before close the producer since the send is async

	producerOptions := stream.NewProducerOptions()
	producerOptions.SetProducerName("producer")

	// Batch 100 Events in the same Frame, and the SDK will handle everything
	// DEDPULICATION DOES NOT WORK WITH SUBENTRY
	producerOptions.SetSubEntrySize(100)
	producerOptions.SetCompression(stream.Compression{}.Gzip())

	producer, err := env.NewProducer(EVENTSTREAM, producerOptions)
	if err != nil {
		panic(err)
	}
	chPublishConfirm := producer.NotifyPublishConfirmation()
	totalConfirmed := int32(0)
	chFinished := make(chan struct{})
	go func() {
		for confirmed := range chPublishConfirm {
			for _, msg := range confirmed {
				if msg.IsConfirmed() {
					atomic.AddInt32(&totalConfirmed, 1)
					if atomic.LoadInt32(&totalConfirmed) == 6001 {
						chFinished <- struct{}{}
					}
				} else {
					fmt.Printf("message %s failed \n  ", msg.GetMessage().GetData())
				}

			}
		}
	}()

	// Publish 6001 messages
	for i := 0; i <= 6001; i++ {
		// fmt.Println("i = ", i)
		event := Event{
			Name: "test",
		}

		data, err := json.Marshal(event)
		if err != nil {
			panic(err)
		}

		message := amqp.NewMessage(data)
		// Apply properties to our message
		props := &amqp.MessageProperties{
			CorrelationID: uuid.NewString(),
		}
		message.Properties = props
		// Set Publishing ID to prevent Deduplication
		message.SetPublishingId(int64(i))
		// Sending the message
		if err := producer.Send(message); err != nil {
			panic(err)
		}
	}

	<-chFinished

	producer.Close()
	env.Close()
	fmt.Printf("Closed with total Confirmed: %d\n", totalConfirmed)

Thanks so much!

Now I run main.go on local system, but the rabbitmq server is on remote, I have a problem when NewProducer, the log is: "panic: Authentication Failure"

But I run main.go on the remote system which the rabbitmq server on , it runs ok.

LucasZhanye · 2024-07-30T07:49:06Z

The debug log is:

2024/07/30 15:48:10 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 15:48:10 [debug] - User root, connected to: 10.2.3.200:5552, vhost:test
2024/07/30 15:48:10 [debug] - Read connection failed: read tcp 172.24.127.40:36958->10.2.3.200:5552: use of closed network connection
2024/07/30 15:48:10 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 15:48:10 [debug] - User root, connected to: 10.2.3.200:5552, vhost:test
2024/07/30 15:48:10 [debug] - Read connection failed: read tcp 172.24.127.40:36960->10.2.3.200:5552: use of closed network connection
2024/07/30 15:48:10 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 15:48:10 [debug] - User root, connected to: 10.2.3.200:5552, vhost:test
2024/07/30 15:48:10 [debug] - Read connection failed: EOF
2024/07/30 15:48:10 [debug] - error removing heartbeat: Response #{id} not found
2024/07/30 15:48:10 [debug] - User:root, Authentication Failure
2024/07/30 15:48:10 [debug] - Read connection failed: read tcp 172.24.127.40:36964->10.2.3.200:5552: use of closed network connection

Gsantomaggio · 2024-07-30T07:51:54Z

2024/07/30 15:48:10 [debug] - User:root, Authentication Failure

I think it is clear where is the problem here

Signed-off-by: Gabriele Santomaggio <[email protected]>

hiimjako · 2024-07-30T08:22:50Z

In my opinion, the problem of connecting to pre-3.11 servers is solved, as confirmed above. If there are no other technical problems, I think it is mergeable

LucasZhanye · 2024-07-30T08:29:21Z

Authentication Failure

Hi @Gsantomaggio Maybe I know why it is the queue is zero.

Because I have two rabbitmq server run by docker container, the first is listen on 5552, and the second is listen on 15552

When I exec main.go which is connect to the port 15552, it will send msg to the rabbitmq server which is listen on 5552.

From the log, at last it use 5552:

2024/07/30 16:22:50 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 16:22:50 [debug] - User root, connected to: 10.2.3.200:15552, vhost:test
2024/07/30 16:22:50 [debug] - Read connection failed: read tcp 10.2.3.200:35938->10.2.3.200:15552: use of closed network connection
2024/07/30 16:22:50 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 16:22:50 [debug] - **User root, connected to: 10.2.3.200:15552**, vhost:test
2024/07/30 16:22:50 [debug] - Read connection failed: read tcp 10.2.3.200:35940->10.2.3.200:15552: use of closed network connection
2024/07/30 16:22:50 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 16:22:50 [debug] - User root, connected to: 10.2.3.200:15552, vhost:test
coordinatorKey =  **localhost:5552**
2024/07/30 16:22:50 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 16:22:50 [debug] - User root, connected to: localhost:5552, vhost:test
2024/07/30 16:22:50 [debug] - Read connection failed: read tcp 10.2.3.200:35942->10.2.3.200:15552: use of closed network connection
2024/07/30 16:22:50 [debug] - waitForInflightMessages, channel: 2374 - pending messages len: 28 - unconfirmed len: 2502 - retry: 0
2024/07/30 16:22:50 [debug] - **Read connection failed: read tcp 127.0.0.1:45690->127.0.0.1:5552**: use of closed network connection
Closed with total Confirmed: 6001

Finally, I see the rabbitmq management which is listen on 5552 , the queue message is increased.

Gsantomaggio · 2024-07-30T08:29:26Z

@hiimjako @LucasZhanye merged.

LucasZhanye · 2024-07-30T08:54:52Z

Authentication Failure

Hi @Gsantomaggio Maybe I know why it is the queue is zero.

Because I have two rabbitmq server run by docker container, the first is listen on 5552, and the second is listen on 15552

When I exec main.go which is connect to the port 15552, it will send msg to the rabbitmq server which is listen on 5552.

From the log, at last it use 5552:

2024/07/30 16:22:50 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 16:22:50 [debug] - User root, connected to: 10.2.3.200:15552, vhost:test
2024/07/30 16:22:50 [debug] - Read connection failed: read tcp 10.2.3.200:35938->10.2.3.200:15552: use of closed network connection
2024/07/30 16:22:50 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 16:22:50 [debug] - **User root, connected to: 10.2.3.200:15552**, vhost:test
2024/07/30 16:22:50 [debug] - Read connection failed: read tcp 10.2.3.200:35940->10.2.3.200:15552: use of closed network connection
2024/07/30 16:22:50 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 16:22:50 [debug] - User root, connected to: 10.2.3.200:15552, vhost:test
coordinatorKey =  **localhost:5552**
2024/07/30 16:22:50 [info] - Server version is less than 3.11.0, skipping command version exchange
2024/07/30 16:22:50 [debug] - User root, connected to: localhost:5552, vhost:test
2024/07/30 16:22:50 [debug] - Read connection failed: read tcp 10.2.3.200:35942->10.2.3.200:15552: use of closed network connection
2024/07/30 16:22:50 [debug] - waitForInflightMessages, channel: 2374 - pending messages len: 28 - unconfirmed len: 2502 - retry: 0
2024/07/30 16:22:50 [debug] - **Read connection failed: read tcp 127.0.0.1:45690->127.0.0.1:5552**: use of closed network connection
Closed with total Confirmed: 6001

Finally, I see the rabbitmq management which is listen on 5552 , the queue message is increased.

I find the solution on #315

I change my code to below:

addressResolver := stream.AddressResolver{
		Host: "10.2.3.200",
		Port: 15552,
	}

	env, err := stream.NewEnvironment(
		stream.NewEnvironmentOptions().
			SetHost(addressResolver.Host).
			SetPort(addressResolver.Port).
			SetAddressResolver(addressResolver).
			SetUser("root").SetPassword("root").SetVHost("test").
			SetMaxProducersPerClient(5))

It works ok!

Gsantomaggio · 2024-08-08T08:55:20Z

@LucasZhanye FYI: https://github.com/rabbitmq/rabbitmq-stream-go-client/releases/tag/v1.4.8

Gsantomaggio added 2 commits July 29, 2024 11:59

Check version when version is not

17dd299

and is less than 3.11 Signed-off-by: Gabriele Santomaggio <[email protected]>

Check version when version is not

ed1606e

and is less than 3.11 Signed-off-by: Gabriele Santomaggio <[email protected]>

Gsantomaggio mentioned this pull request Jul 29, 2024

Can't not connect to Rabbitmq Stream via port 5552 #344

Closed

Gsantomaggio marked this pull request as ready for review July 29, 2024 11:51

Reduce the log level

2d7a67e

Signed-off-by: Gabriele Santomaggio <[email protected]>

hiimjako added the bug Something isn't working label Jul 30, 2024

Gsantomaggio merged commit e0bcf7e into main Jul 30, 2024
4 checks passed

Gsantomaggio deleted the check_version branch July 30, 2024 08:29

Gsantomaggio changed the title ~~Check version when version is not~~ Check version when serverProperties is not empty and < 3.11.0 Jul 30, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Check version when serverProperties is not empty and < 3.11.0 #345

Check version when serverProperties is not empty and < 3.11.0 #345

Gsantomaggio commented Jul 29, 2024 •

edited

Loading

Gsantomaggio commented Jul 29, 2024

LucasZhanye commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

Gsantomaggio commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

Gsantomaggio commented Jul 30, 2024

hiimjako commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024 •

edited

Loading

Gsantomaggio commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

Gsantomaggio commented Aug 8, 2024

Check version when serverProperties is not empty and < 3.11.0 #345

Check version when serverProperties is not empty and < 3.11.0 #345

Conversation

Gsantomaggio commented Jul 29, 2024 • edited Loading

Gsantomaggio commented Jul 29, 2024

LucasZhanye commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

Gsantomaggio commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

Gsantomaggio commented Jul 30, 2024

hiimjako commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024 • edited Loading

Gsantomaggio commented Jul 30, 2024

LucasZhanye commented Jul 30, 2024

Gsantomaggio commented Aug 8, 2024

Gsantomaggio commented Jul 29, 2024 •

edited

Loading

LucasZhanye commented Jul 30, 2024 •

edited

Loading