Repository: mateusjunges/laravel-kafka (php)

Issue:
[BUG] CONFWARN on consumer with auto.offset.reset
**Describe the bug**
CONFWARN occurs when starting $consumer->consume();

**To Reproduce**
Steps to reproduce the behavior:
1. Start consumer with option `auto.offset.reset` 
```php 
 $consumer = Kafka::createConsumer(['topic', 'topic2'], 'group1', $broker_addr)
            ->withOptions([
                'auto.offset.reset'                     => 'earliest'
            ])
            ->withHandler(function (ConsumedMessage $message) { })
            ->build();
 $consumer->consume();
```
2. We get the warning "Configuration property auto.offset.reset is a consumer property and will be ignored by this producer instance". But we did not launch a producer, only a consumer
3. We get the warning "Configuration property compression.codec is a producer property and will be ignored by this consumer instance". But we did not specify the option `compression.codec` 

Why is the option `compression.codec` passed to `getConsumerOptions` , which is relevant only for the producer
https://github.com/mateusjunges/laravel-kafka/blob/b11aa8fb78f009b95e915eead4b5c5aadb732117/src/Config/Config.php#L70

Why is a producer created when calling `consume()`?
https://github.com/mateusjunges/laravel-kafka/blob/b11aa8fb78f009b95e915eead4b5c5aadb732117/src/Consumers/Consumer.php#L78

**Expected behavior**
The consumer will start without errors or warnings.

**Additional context**
Logs:

```log
%4|1645017413.959|CONFWARN|rdkafka#consumer-1| [thrd:app]: Configuration property compression.codec is a producer property and will be ignored by this consumer instance
%4|1645017413.963|CONFWARN|rdkafka#producer-2| [thrd:app]: Configuration property auto.offset.reset is a consumer property and will be ignored by this producer instance
```

Pull request:
Remove producer config property from `getConsumerOptions` method
This PR removes the `compression.codec` configuration option from the consumer configuration. Fixes #75 

Patch:
diff --git a/src/Config/Config.php b/src/Config/Config.php
index 1610803..a057e62 100644
--- a/src/Config/Config.php
+++ b/src/Config/Config.php
@@ -9,6 +9,55 @@ class Config
 {
     const SASL_PLAINTEXT = 'SASL_PLAINTEXT';
     const SASL_SSL = 'SASL_SSL';
+    const PRODUCER_ONLY_CONFIG_OPTIONS = [
+        'transactional.id',
+        'transaction.timeout.ms',
+        'enable.idempotence',
+        'enable.gapless.guarantee',
+        'queue.buffering.max.messages',
+        'queue.buffering.max.kbytes',
+        'queue.buffering.max.ms',
+        'linger.ms',
+        'message.send.max.retries',
+        'retries',
+        'retry.backoff.ms',
+        'queue.buffering.backpressure.threshold',
+        'compression.codec',
+        'compression.type',
+        'batch.num.messages',
+        'batch.size',
+        'delivery.report.only.error',
+        'dr_cb',
+        'dr_msg_cb',
+        'sticky.partitioning.linger.ms'
+    ];
+    const CONSUMER_ONLY_CONFIG_OPTIONS = [
+        'partition.assignment.strategy',
+        'session.timeout.ms',
+        'heartbeat.interval.ms',
+        'group.protocol.type',
+        'coordinator.query.interval.ms',
+        'max.poll.interval.ms',
+        'enable.auto.commit',
+        'auto.commit.interval.ms',
+        'enable.auto.offset.store',
+        'queued.min.messages',
+        'queued.max.messages.kbytes',
+        'fetch.wait.max.ms',
+        'fetch.message.max.bytes',
+        'max.partition.fetch.bytes',
+        'fetch.max.bytes',
+        'fetch.min.bytes',
+        'fetch.error.backoff.ms',
+        'offset.store.method',
+        'isolation.level',
+        'consume_cb',
+        'rebalance_cb',
+        'offset_commit_cb',
+        'enable.partition.eof',
+        'check.crcs',
+        'allow.auto.create.topics'
+    ];
 
     public function __construct(
         private string $broker,
@@ -67,7 +116,6 @@ class Config
             'metadata.broker.list' => $this->broker,
             'auto.offset.reset' => config('kafka.offset_reset', 'latest'),
             'enable.auto.commit' => config('kafka.auto_commit', true) === true ? 'true' : 'false',
-            'compression.codec' => config('kafka.compression', 'snappy'),
             'group.id' => $this->groupId,
             'bootstrap.servers' => $this->broker,
         ];
@@ -76,7 +124,9 @@ class Config
             $options['enable.auto.commit'] = $this->autoCommit === true ? 'true' : 'false';
         }
 
-        return array_merge($options, $this->customOptions, $this->getSaslOptions());
+        return collect(array_merge($options, $this->customOptions, $this->getSaslOptions()))
+            ->reject(fn ($option) => in_array($option, self::PRODUCER_ONLY_CONFIG_OPTIONS))
+            ->toArray();
     }
 
     #[Pure]
@@ -88,7 +138,9 @@ class Config
             'metadata.broker.list' => $this->broker,
         ];
 
-        return array_merge($config, $this->customOptions, $this->getSaslOptions());
+        return collect(array_merge($config, $this->customOptions, $this->getSaslOptions()))
+            ->reject(fn ($option) => in_array($option, self::CONSUMER_ONLY_CONFIG_OPTIONS))
+            ->toArray();
     }
 
     #[Pure]


Test patch:
diff --git a/tests/Config/ConfigTest.php b/tests/Config/ConfigTest.php
index fc9dfd1..8e3838d 100644
--- a/tests/Config/ConfigTest.php
+++ b/tests/Config/ConfigTest.php
@@ -25,7 +25,6 @@ class ConfigTest extends LaravelKafkaTestCase
         $expectedOptions = [
             'auto.offset.reset' => 'latest',
             'enable.auto.commit' => 'true',
-            'compression.codec' => 'snappy',
             'group.id' => 'group',
             'bootstrap.servers' => 'broker',
             'metadata.broker.list' => 'broker',