@Test
public void test() throws Exception {

  Schema schema = SchemaBuilder.record("r").fields().name("f").type().stringType().noDefault().endRecord();
  Record value = new Record(schema);
  ConsumerRecord<Void, Record> record1 = new ConsumerRecord<>(topic, 0, 1, null, value);
  ConsumerRecords<Void, Record> records1 = new ConsumerRecords<>(
      ImmutableMap.of(partition, ImmutableList.of(record1)));
  when(consumer.poll(pollTimeout)).thenReturn(records1);
  underTest.run(null);
  InOrder inOrder = inOrder(consumer, writer, context);
  inOrder.verify(consumer).assign(partitions);
  inOrder.verify(consumer).seek(partition, 1L);
  inOrder.verify(consumer).poll(pollTimeout);
  inOrder.verify(writer).write(record1);
  inOrder.verify(consumer).pause(partitions);
  inOrder.verify(writer).close();
  inOrder.verify(context).close();
}