Move auto-commit from subscription to consumer

Fixes a librdkafka deprecation warning:

Configuration property auto.commit.enable is deprecated: [**LEGACY PROPERTY:** This property is used by the simple legacy consumer only. When using the high-level KafkaConsumer, the global `enable.auto.commit` property must be used instead]. If true, periodically commit offset of the last message handed to the application. This committed offset will be used when the process restarts to pick up where it left off. If false, the application will have to call `rd_kafka_offset_store()` to store an offset (optional). **NOTE:** There is currently no zookeeper integration, offsets will be written to broker or local file according to offset.store.method.

Author: Philipp Hausmann

src/Kafka/Consumer/ConsumerProperties.hs

@@ -4,6 +4,7 @@ module Kafka.Consumer.ConsumerProperties
 ( ConsumerProperties(..)
 , CallbackMode(..)
 , brokersList
+, autoCommit
 , noAutoCommit
 , noAutoOffsetStore
 , groupId
@@ -29,7 +30,7 @@ import           Data.Text            (Text)
 import qualified Data.Text            as Text
 import           Kafka.Consumer.Types (ConsumerGroupId (..))
 import           Kafka.Internal.Setup (KafkaConf (..))
-import           Kafka.Types          (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), kafkaCompressionCodecToText, kafkaDebugToText)
+import           Kafka.Types          (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), kafkaCompressionCodecToText, kafkaDebugToText, Millis(..))
 
 import Kafka.Consumer.Callbacks as X
 
@@ -65,6 +66,13 @@ brokersList bs =
   let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs)
    in extraProps $ M.fromList [("bootstrap.servers", bs')]
 
+autoCommit :: Millis -> ConsumerProperties
+autoCommit (Millis ms) = extraProps $
+  M.fromList
+    [ ("enable.auto.commit", "true")
+    , ("auto.commit.interval.ms", Text.pack $ show ms)
+    ]
+
 -- | Disables auto commit for the consumer
 noAutoCommit :: ConsumerProperties
 noAutoCommit =

src/Kafka/Consumer/Subscription.hs

@@ -4,7 +4,6 @@ module Kafka.Consumer.Subscription
 ( Subscription(..)
 , topics
 , offsetReset
-, autoCommit
 , extraSubscriptionProps
 )
 where
@@ -44,12 +43,5 @@ offsetReset o =
              Latest   -> "latest"
    in Subscription (Set.empty) (M.fromList [("auto.offset.reset", o')])
 
-autoCommit :: Millis -> Subscription
-autoCommit (Millis ms) = Subscription (Set.empty) $
-  M.fromList
-    [ ("enable.auto.commit", "true")
-    , ("auto.commit.interval.ms", Text.pack $ show ms)
-    ]
-
 extraSubscriptionProps :: Map Text Text -> Subscription
 extraSubscriptionProps = Subscription (Set.empty)