commit method

@override Future<void> commit(String topic, int offset)

Commit offset as consumed, so a restart resumes after it.

Implementation

@override
Future<void> commit(String topic, int offset) async {
  await _findCoordinator();
  final _Writer body = _Writer()
    ..string(group)
    ..int32(-1) // generation: a simple consumer, not a group member
    ..string('') // member id, empty for the same reason
    ..int64(-1) // retention: the broker's default
    ..int32(1)
    ..string(topic)
    ..int32(1)
    ..int32(0)
    ..int64(offset)
    ..string(''); // metadata

  final _Response response = await _call(8, 2, body.bytes); // OffsetCommit v2
  final _Reader read = _Reader(response.body);
  read.int32();
  read.string();
  read.int32();
  read.int32();
  final int error = read.int16();
  if (error != 0) {
    throw StateError('Kafka refused the offset commit with error $error.');
  }
}