Removes improper configs

Some producer configs were being passed to the consumer and vice versa.
Such improper configs were causing some configuration warnings. This
commit fixes the configs and removes the problem causing the warnings.

Author: joesantos418

dev/Dockerfile

@@ -1,4 +1,4 @@
-FROM joesantos386/laravel:php7.4-v1.5.2-4.0.4-8
+FROM joesantos386/laravel:8.0-v1.5.3-5.0.0-8
 
 RUN apk add libzip-dev
 

src/Consumer.php

@@ -48,8 +48,8 @@ public function __construct(Config $config)
 
     public function consume(): void
     {
-        $this->consumer = new KafkaConsumer($this->setConf());
-        $this->producer = new Producer($this->setConf());
+        $this->consumer = new KafkaConsumer($this->setConsumerConf());
+        $this->producer = new Producer($this->setProducerConf());
 
         $this->committer = CommitterBuilder::withConsumer($this->consumer)
             ->andRetry(new NativeSleeper(), $this->config->getMaxCommitRetries())
@@ -71,13 +71,12 @@ private function doConsume()
         $this->handleMessage($message);
     }
 
-    private function setConf(): Conf
+    private function setConsumerConf(): Conf
     {
         $conf = new Conf();
         $conf->set('auto.offset.reset', 'smallest');
         $conf->set('queued.max.messages.kbytes', '10000');
         $conf->set('enable.auto.commit', 'false');
-        $conf->set('compression.codec', 'gzip');
         $conf->set('max.poll.interval.ms', '86400000');
         $conf->set('group.id', $this->config->getGroupId());
         $conf->set('bootstrap.servers', $this->config->getBroker());
@@ -92,6 +91,22 @@ private function setConf(): Conf
         return $conf;
     }
 
+    private function setProducerConf(): Conf
+    {
+        $conf = new Conf();
+        $conf->set('compression.codec', 'gzip');
+        $conf->set('bootstrap.servers', $this->config->getBroker());
+        $conf->set('security.protocol', $this->config->getSecurityProtocol());
+
+        if ($this->config->isPlainText() && $this->config->getSasl() !== null) {
+            $conf->set('sasl.username', $this->config->getSasl()->getUsername());
+            $conf->set('sasl.password', $this->config->getSasl()->getPassword());
+            $conf->set('sasl.mechanisms', $this->config->getSasl()->getMechanisms());
+        }
+
+        return $conf;
+    }
+
     private function executeMessage(Message $message): void
     {
         try {