Skip to content

Commit

Permalink
Added convenience method StoreMessage that calls StoreOffsets (@finnc…
Browse files Browse the repository at this point in the history
…olman, #676)

* Added convenience method StoreMessage that calls StoreOffsets

* Added '.' at the end of the comment

* Added Offset less than 0 check

* Moved the check to the correct method

Co-authored-by: Finn Colman <finn.colman@modicagroup.com>
  • Loading branch information
finncolman and finncolman committed Oct 28, 2021
1 parent 7182357 commit 194896a
Show file tree
Hide file tree
Showing 2 changed files with 24 additions and 0 deletions.
14 changes: 14 additions & 0 deletions kafka/consumer.go
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,20 @@ func (c *Consumer) StoreOffsets(offsets []TopicPartition) (storedOffsets []Topic
return storedOffsets, nil
}

// StoreMessage stores offset based on the provided message.
// This is a convenience method that uses StoreOffsets to do the actual work.
func (c *Consumer) StoreMessage(m *Message) (storedOffsets []TopicPartition, err error) {
if m.TopicPartition.Error != nil {
return nil, newErrorFromString(ErrInvalidArg, "Can't store errored message")
}
if m.TopicPartition.Offset < 0 {
return nil, newErrorFromString(ErrInvalidArg, "Can't store message with offset less than 0")
}
offsets := []TopicPartition{m.TopicPartition}
offsets[0].Offset++
return c.StoreOffsets(offsets)
}

// Seek seeks the given topic partitions using the offset from the TopicPartition.
//
// If timeoutMs is not 0 the call will wait this long for the
Expand Down
10 changes: 10 additions & 0 deletions kafka/consumer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,16 @@ func TestConsumerAPIs(t *testing.T) {
t.Errorf("StoreOffsets(empty) failed: %s", err)
}

// test StoreMessage doesn't fail either
stored, err = c.StoreMessage(&Message{TopicPartition: TopicPartition{Topic: &topic, Partition: 0, Offset: 1}})
if err != nil && err.(Error).Code() != ErrUnknownPartition {
t.Errorf("StoreMessage() failed: %s", err)
toppar := stored[0]
if toppar.Error != nil && toppar.Error.(Error).Code() == ErrUnknownPartition {
t.Errorf("StoreMessage() TopicPartition error: %s", toppar.Error)
}
}

topic1 := "gotest1"
topic2 := "gotest2"
err = c.Assign([]TopicPartition{{Topic: &topic1, Partition: 2},
Expand Down

0 comments on commit 194896a

Please sign in to comment.