def restore(self, *, topic):
    topic_partition_files = self.list_topic_data_files(topic=topic)
    partition_offset_records = {}
    since = self.config.get('since')
    with TemporaryDirectory() as working_directory:
        while True:
            progress = False
            for partition in topic_partition_files:
                if topic_partition_files[partition]:
                    object_record = topic_partition_files[partition][0]
                    topic_partition_files[partition] = topic_partition_files[partition][1:]
                    progress = True
                    object_name = object_record['object_name']
                    if ((since is not None) and (since > object_record['last_modified'])):
                        self.log.info('Skipping object %r due to timestamp', object_name)
                        continue
                    local_name = f'{working_directory}/{topic}-{partition}'
                    self.object_storage.get_contents_to_file(object_name, local_name)
                    if (object_record.get('compression') == 'gzip'):
                        fh = gzip.open(local_name, 'rt')
                    else:
                        fh = open(local_name, 'r')
                    nrecords = 0
                    for line in fh.readlines():
                        (key, value, offset, timestamp) = self.parse_record(line.strip())
                        future_record = self.kafka_producer.send(topic, partition=partition, key=key, value=value, timestamp_ms=timestamp)
                        nrecords += 1
                        partition_offset_records[partition] = {'last_original_offset': offset, 'last_produced_record': future_record}
                    self.log.info('Restored %d messages from object %r', nrecords, object_name)
                    fh.close()
                    os.unlink(local_name)
            if (not progress):
                self.kafka_producer.flush()
                break
    for partition in sorted(partition_offset_records):
        self.log.info('Partition %d original offset %d new offset %d', partition, partition_offset_records[partition]['last_original_offset'], partition_offset_records[partition]['last_produced_record'].get().offset)
def restore(self, *, topic):
    topic_partition_files = self.list_topic_data_files(topic=topic)
    partition_offset_records = {}
    since = self.config.get('since')
    with TemporaryDirectory() as working_directory:
        while True:
            progress = False
            for partition in topic_partition_files:
                if topic_partition_files[partition]:
                    object_record = topic_partition_files[partition][0]
                    topic_partition_files[partition] = topic_partition_files[partition][1:]
                    progress = True
                    object_name = object_record['object_name']
                    if ((since is not None) and (since > object_record['last_modified'])):
                        self.log.info('Skipping object %r due to timestamp', object_name)
                        continue
                    local_name = f'{working_directory}/{topic}-{partition}'
                    self.object_storage.get_contents_to_file(object_name, local_name)
                    if (object_record.get('compression') == 'gzip'):
                        fh = gzip.open(local_name, 'rt')
                    else:
                        fh = open(local_name, 'r')
                    nrecords = 0
                    for line in fh.readlines():
                        (key, value, offset, timestamp) = self.parse_record(line.strip())
                        future_record = self.kafka_producer.send(topic, partition=partition, key=key, value=value, timestamp_ms=timestamp)
                        nrecords += 1
                        partition_offset_records[partition] = {'last_original_offset': offset, 'last_produced_record': future_record}
                    self.log.info('Restored %d messages from object %r', nrecords, object_name)
                    fh.close()
                    os.unlink(local_name)
            if (not progress):
                self.kafka_producer.flush()
                break
    for partition in sorted(partition_offset_records):
        self.log.info('Partition %d original offset %d new offset %d', partition, partition_offset_records[partition]['last_original_offset'], partition_offset_records[partition]['last_produced_record'].get().offset)