def __init__(self, *, config):
    self.log = logging.getLogger(self.__class__.__name__)
    self.config = config
    object_storage_config = self.config.get('object_storage', {})
    object_storage_type = object_storage_config.get('type')
    if (object_storage_type == 'gcs'):
        self.object_storage = GCSProvider(config=object_storage_config)
    else:
        raise ValueError(f'Unknown object storage type: {object_storage_type}')
    kafka_config = self.config.get('kafka', {})
    if (('ssl_cafile' in kafka_config) and ('ssl_access_certificate_file' in kafka_config) and ('ssl_access_key_file' in kafka_config)):
        self.kafka_producer = kafka.KafkaProducer(bootstrap_servers=kafka_config['kafka_url'], security_protocol='SSL', ssl_cafile=kafka_config['ssl_ca_file'], ssl_certfile=kafka_config['ssl_access_certificate_file'], ssl_keyfile=kafka_config['ssl_access_key_file'])
    else:
        self.kafka_producer = kafka.KafkaProducer(bootstrap_servers=kafka_config['kafka_url'])
def __init__(self, *, config):
    self.log = logging.getLogger(self.__class__.__name__)
    self.config = config
    object_storage_config = self.config.get('object_storage', {})
    object_storage_type = object_storage_config.get('type')
    if (object_storage_type == 'gcs'):
        self.object_storage = GCSProvider(config=object_storage_config)
    else:
        raise ValueError(f'Unknown object storage type: {object_storage_type}')
    kafka_config = self.config.get('kafka', {})
    if (('ssl_cafile' in kafka_config) and ('ssl_access_certificate_file' in kafka_config) and ('ssl_access_key_file' in kafka_config)):
        self.kafka_producer = kafka.KafkaProducer(bootstrap_servers=kafka_config['kafka_url'], security_protocol='SSL', ssl_cafile=kafka_config['ssl_ca_file'], ssl_certfile=kafka_config['ssl_access_certificate_file'], ssl_keyfile=kafka_config['ssl_access_key_file'])
    else:
        self.kafka_producer = kafka.KafkaProducer(bootstrap_servers=kafka_config['kafka_url'])