feat: Expose Pause and Resume through kafka.Consumer interface (#230)

Author: Kelsey

kafka/consume.go

@@ -47,6 +47,12 @@ type Consumer interface {
 
 	// SeekToTime seeks to the specified time.
 	SeekToTime(t time.Time) error
+
+	// Pause pauses consumption of the provided partitions
+	Pause(p []kafkalib.TopicPartition) error
+
+	// Resume resumes consumption of the provided partitions
+	Resume(p []kafkalib.TopicPartition) error
 }
 
 // ConfluentConsumer implements Consumer interface.
@@ -331,6 +337,18 @@ func (cc *ConfluentConsumer) CommitMessage(msg *kafkalib.Message) error {
 	return errors.Wrap(err, "failed committing Kafka message")
 }
 
+// Pause pauses consumption of the provided partitions
+func (cc *ConfluentConsumer) Pause(p []kafkalib.TopicPartition) error {
+	err := cc.c.Pause(p)
+	return errors.Wrap(err, "failed to pause Kafka topic")
+}
+
+// Resume resumes consumption of the provided partitions
+func (cc *ConfluentConsumer) Resume(p []kafkalib.TopicPartition) error {
+	err := cc.c.Resume(p)
+	return errors.Wrap(err, "failed to resume Kafka topic")
+}
+
 // Close closes the consumer.
 func (cc *ConfluentConsumer) Close() error {
 	return cc.c.Close()

kafka/integration_test.go

@@ -92,6 +92,9 @@ func TestIntegration(t *testing.T) {
 			assert.Equal([]byte(k), m.Key, "Partition to read from: %d, Msg: %+v", p, m)
 			assert.Equal([]byte(v), m.Value, "Partition to read from: %d, Msg: %+v", p, m)
 
+			assert.NoError(c.Pause([]kafkalib.TopicPartition{m.TopicPartition}))
+			assert.NoError(c.Resume([]kafkalib.TopicPartition{m.TopicPartition}))
+
 			assert.NoError(c.CommitMessage(m))
 		}
 

kafka/kafkatest/utils.go

@@ -156,6 +156,14 @@ func (f *FakeKafkaConsumer) SeekToTime(t time.Time) error {
 	return nil // noop
 }
 
+func (f *FakeKafkaConsumer) Pause(p []kafkalib.TopicPartition) error {
+	return nil // noop
+}
+
+func (f *FakeKafkaConsumer) Resume(p []kafkalib.TopicPartition) error {
+	return nil // noop
+}
+
 func (f *FakeKafkaConsumer) Close() error {
 	close(f.commits)
 	return nil