-
Notifications
You must be signed in to change notification settings - Fork 2
Fixed some functional issues and improved code readability #3
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
c084c26
aecb741
8763ca2
78edc63
74d292b
6b2f2de
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -3,12 +3,13 @@ package internal_test | |
import ( | ||
"context" | ||
"errors" | ||
"testing" | ||
"time" | ||
|
||
"github.com/IBM/sarama" | ||
"github.com/stretchr/testify/assert" | ||
"github.com/violetpay-org/queue-streamer/common" | ||
"github.com/violetpay-org/queue-streamer/internal" | ||
"testing" | ||
"time" | ||
) | ||
|
||
var cbrokers = []string{"localhost:9093"} | ||
|
@@ -136,6 +137,7 @@ func TestStreamConsumer_ConsumeClaim(t *testing.T) { | |
msg.DataChan = make(chan *sarama.ConsumerMessage, 1) | ||
|
||
go func() { | ||
defer close(msg.DataChan) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Closing the channel is used as a signal that the channel is no longer in use. It is not related to memory leaks. So, 'close' should be called when the data transfer ends, rather than when the goroutine ends (defer). There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. In this case deleted the code line (close) below and used the defer instead. |
||
time.Sleep(1 * time.Second) | ||
msg.DataChan <- &sarama.ConsumerMessage{ | ||
Topic: "test", | ||
|
@@ -145,7 +147,6 @@ func TestStreamConsumer_ConsumeClaim(t *testing.T) { | |
Offset: 0, | ||
} | ||
time.Sleep(1 * time.Second) | ||
close(msg.DataChan) | ||
}() | ||
|
||
assert.Equal(t, 0, len(consumer.ProducerPool().Producers())) | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
A transaction must be canceled if some error occurs, for exactly-once. So this is good for infinite loop, but this doesn't look good for business policy. In my opinion, it would be better to delay processing (time.Sleep) or log if cross the attempt threshold in a loop. (actually my suggestion is another business feature)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If a transaction error occurs, it will one day be processed if the producer is okay, so no worry to infinite loop. If there is a problem with the producer, it is prevented by handling
producer.TxnStatus()& sarama.ProducerTxnFlagFatalError