master
This commit is contained in:
+2
-52
@@ -15,7 +15,7 @@
|
|||||||
|
|
||||||
REFERENCES:
|
REFERENCES:
|
||||||
|
|
||||||
N/A
|
01. https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#
|
||||||
|
|
||||||
DOWNLOADS:
|
DOWNLOADS:
|
||||||
|
|
||||||
@@ -50,9 +50,6 @@ from icecream import IceCreamDebugger
|
|||||||
# To work with datatypes:
|
# To work with datatypes:
|
||||||
from typing import List, Literal
|
from typing import List, Literal
|
||||||
|
|
||||||
# For SSL security:
|
|
||||||
import ssl
|
|
||||||
|
|
||||||
|
|
||||||
# *****************************************************************************************************************
|
# *****************************************************************************************************************
|
||||||
# ***** ****
|
# ***** ****
|
||||||
@@ -81,54 +78,7 @@ import ssl
|
|||||||
# *****************************************************************************************************************
|
# *****************************************************************************************************************
|
||||||
|
|
||||||
|
|
||||||
def create_producer_config(
|
# --- Nothing Yet
|
||||||
bootstrap_servers: str | List[str],
|
|
||||||
group_id: str | None = None,
|
|
||||||
security_protocol: Literal["PLAINTEXT", "SSL"] = "PLAINTEXT",
|
|
||||||
ca_file: str | None = None,
|
|
||||||
cert_file: str | None = None,
|
|
||||||
key_file: str | None = None,
|
|
||||||
client_id: str | None = None,
|
|
||||||
buffer_memory: int = None
|
|
||||||
) -> dict:
|
|
||||||
|
|
||||||
"""
|
|
||||||
Creates the config required for Confluent-Kafka's library.
|
|
||||||
:param bootstrap_servers: The addresses of the Kafka brokers.
|
|
||||||
:param group_id: The identifier of the group of consumers. Irrelevant for producers.
|
|
||||||
:param security_protocol: What sort of security protocol to use.
|
|
||||||
:param ca_file: Needed for 'SSL' security protocol.
|
|
||||||
:param cert_file: Needed for 'SSL' security protocol.
|
|
||||||
:param key_file: Needed for 'SSL' security protocol.
|
|
||||||
:param client_id: An identifier for one producer/consumer. Useful for debugging later.
|
|
||||||
:param buffer_memory: The size of the local message buffer in bytes (only for producers).
|
|
||||||
:return: The dictionary that needs to be passed as the 'conf' param when creating the producer/consumer.
|
|
||||||
"""
|
|
||||||
|
|
||||||
# Start with the bare minimum:
|
|
||||||
if not isinstance(bootstrap_servers, list): bootstrap_servers = [bootstrap_servers]
|
|
||||||
config = {
|
|
||||||
"bootstrap.servers": ",".join(bootstrap_servers),
|
|
||||||
"security.protocol": security_protocol
|
|
||||||
}
|
|
||||||
|
|
||||||
# Add the SSL security details:
|
|
||||||
if security_protocol == "SSL":
|
|
||||||
config["ssl.ca.location"] = ca_file
|
|
||||||
config["ssl.certificate.location"] = cert_file
|
|
||||||
config["ssl.key.location"] = key_file
|
|
||||||
|
|
||||||
# Specific to consumers:
|
|
||||||
if group_id: config["group.id"] = group_id
|
|
||||||
|
|
||||||
# Specific to producers:
|
|
||||||
if buffer_memory: config["buffer.memory"] = buffer_memory
|
|
||||||
|
|
||||||
# Add an identifier for debugging:
|
|
||||||
if client_id: config["client.id"] = client_id
|
|
||||||
|
|
||||||
# Done here:
|
|
||||||
return config
|
|
||||||
|
|
||||||
|
|
||||||
# *****************************************************************************************************************
|
# *****************************************************************************************************************
|
||||||
|
|||||||
Reference in New Issue
Block a user