Skip to content
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.

Sign up for GitHub

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

Permit setting version on mock produce response #999

Merged
merged 2 commits into from
Jan 4, 2018
Merged

Permit setting version on mock produce response #999

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
893be7d
Permit setting version on mock produce response
fgrosse Dec 6, 2017
5169c31
Fix formatting
fgrosse Jan 4, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions async_producer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -639,6 +639,7 @@ func TestAsyncProducerFlusherRetryCondition(t *testing.T) {

leader.SetHandlerByMap(map[string]MockResponse{
"ProduceRequest": NewMockProduceResponse(t).
SetVersion(0).
SetError("my_topic", 0, ErrNoError),
})

Expand Down
14 changes: 11 additions & 3 deletions mockresponses.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -384,14 +384,20 @@ func (mr *MockOffsetCommitResponse) getError(group, topic string, partition int3

// MockProduceResponse is a `ProduceResponse` builder.
type MockProduceResponse struct {
errors map[string]map[int32]KError
t TestReporter
version int16
errors map[string]map[int32]KError
t TestReporter
}

func NewMockProduceResponse(t TestReporter) *MockProduceResponse {
return &MockProduceResponse{t: t}
}

func (mr *MockProduceResponse) SetVersion(version int16) *MockProduceResponse {
mr.version = version
return mr
}

func (mr *MockProduceResponse) SetError(topic string, partition int32, kerror KError) *MockProduceResponse {
if mr.errors == nil {
mr.errors = make(map[string]map[int32]KError)
Expand All @@ -407,7 +413,9 @@ func (mr *MockProduceResponse) SetError(topic string, partition int32, kerror KE

func (mr *MockProduceResponse) For(reqBody versionedDecoder) encoder {
req := reqBody.(*ProduceRequest)
res := &ProduceResponse{}
res := &ProduceResponse{
Version: mr.version,
}
for topic, partitions := range req.records {
for partition := range partitions {
res.AddTopicPartition(topic, partition, mr.getError(topic, partition))
Expand Down