Typo in AlertConsumer code documentation

[AlexGKim]

Two of the config keys given in the AlertConsumer code documentation are incorrect

fink-client/fink_client/consumer.py

Line 42 in 700f579

username: str

username -> sasl.username
password -> sasl.password

[JulienPeloton]

Thanks for spotting the mistake! I will correct it asap.

[JulienPeloton]

Well actually the doc is correct -- AlertConsumer takes as input a dictionnary config whose keys are username and password among several. Then it gets translated to Kafka input parameter names in the _get_kafka_config method:

fink-client/fink_client/consumer.py

Lines 263 to 267 in 4550eca

	if 'username' in config and 'password' in config:
	kafka_config["security.protocol"] = "sasl_plaintext"
	kafka_config["sasl.mechanism"] = "SCRAM-SHA-512"
	kafka_config["sasl.username"] = config["username"]
	kafka_config["sasl.password"] = config["password"]

It is somehow unfortunate that the names are not exactly the same, but the reason behind is that by using username and password on a higher level, it abstracts the authentication method on the lower level (sasl now, but it could be kerberos in the future for example).

I've made the documentation more explicit, see #102

[AlexGKim]

I did pass in a dictionary with "username" and "password" keys but got an error.

[JulienPeloton]

This is surprising. What error exactly did you get? Could you paste the traceback here?

I just tried the following snippet with my credentials, and it worked:

from fink_client.consumer import AlertConsumer

myconfig = {
    'username': '__my_username__',
    'bootstrap.servers': '__servers__',
    'group_id': '__my_group_id__'
}

topics = ["fink_early_sn_candidates_ztf"]

# Instantiate a consumer
consumer = AlertConsumer(topics, myconfig)

# Poll the servers
topic, alert, key = consumer.poll(maxtimeout=5)
print(alert)

[AlexGKim]

Traceback (most recent call last):
  File "manage.py", line 22, in <module>
    main()
  File "manage.py", line 18, in main
    execute_from_command_line(sys.argv)
  File "/Users/akim/venv/tom_env/lib/python3.8/site-packages/django/core/management/__init__.py", line 419, in execute_from_command_line
    utility.execute()
  File "/Users/akim/venv/tom_env/lib/python3.8/site-packages/django/core/management/__init__.py", line 413, in execute
    self.fetch_command(subcommand).run_from_argv(self.argv)
  File "/Users/akim/venv/tom_env/lib/python3.8/site-packages/django/core/management/base.py", line 354, in run_from_argv
    self.execute(*args, **cmd_options)
  File "/Users/akim/venv/tom_env/lib/python3.8/site-packages/django/core/management/base.py", line 398, in execute
    output = self.handle(*args, **options)
  File "/Users/akim/project/tom_desc/stream/management/commands/fink.py", line 55, in handle
    consumer = AlertConsumer(["fink_early_sn_candidates_ztf"], FINK_CONSUMER_CONFIGURATION)
  File "/Users/akim/venv/tom_env/lib/python3.8/site-packages/fink_client/consumer.py", line 55, in __init__
    self._consumer = confluent_kafka.Consumer(self._kafka_config)
cimpl.KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="Failed to create consumer: sasl.username and sasl.password must be set"}

[AlexGKim]

FINK_CONSUMER_CONFIGURATION = {
    # 'sasl.username': os.getenv('FINK_USERNAME',''),
    # 'sasl.password': os.getenv('FINK_PASSWORD', ''),
    'username': os.getenv('FINK_USERNAME',''),
    'password': os.getenv('FINK_PASSWORD', ''),
    'group_id': os.getenv('FINK_GROUP_ID', ''),
    'bootstrap.servers': os.getenv('FINK_SERVER', ''),
    }

[JulienPeloton]

The problem is the password -- if unset (i.e. None in this case), the code will fail because of this line:

fink-client/fink_client/consumer.py

Lines 263 to 267 in 4550eca

	if 'username' in config and 'password' in config:
	kafka_config["security.protocol"] = "sasl_plaintext"
	kafka_config["sasl.mechanism"] = "SCRAM-SHA-512"
	kafka_config["sasl.username"] = config["username"]
	kafka_config["sasl.password"] = config["password"]

I used to not pass it when unset, but I will fix the code as well such that passing None will initialise the consumer as well.

[JulienPeloton]

Hum, actually there might be a breach -- let me check things on the server side...

[AlexGKim]

You are much more computer savvy than I am. But I did notice a change as some point where
key in dictionary no longer worked, had to do key in dictionary.keys()

[JulienPeloton]

@AlexGKim thanks for putting your finger on a larger problem -- I found the authentication mechanism of the cluster was not properly configured (username and password were skipped).

I will fix it, but later. Having said that, this does not prevent data to flow normally, but you will authenticate with the group_id only.