<commit_before>(ns uswitch.bifrost.zk
  (:require [clj-kafka.zk :refer (committed-offset set-offset!)]
            [com.stuartsierra.component :refer (Lifecycle)]
            [clojure.core.async :refer (go-loop <!)]
            [clojure.tools.logging :refer (info error)]))

(defrecord ZookeeperTracker [consumer-properties commit-offset-ch]
  Lifecycle
  (start [this]
    (info "Starting ZookeeperTracker")
    (let [consumer-group-id (consumer-properties "group.id")]
      (go
       (loop [{:keys [topic partition offset] :as m} (<! commit-offset-ch)]
         (when m
           (info "Committing consumer information to ZooKeeper:" m)
           (try
             (set-offset! consumer-properties consumer-group-id topic partition offset)
             (info "Committed offset information to ZooKeeper" m)
             (catch Exception e
               (error e "Unable to commit offset to ZooKeeper")))
           (recur (<! commit-offset-ch))))
       (info "Stopping ZookeeperTracker"))
      this))
  (stop [this]<commit_msg>Fix ZooKeeper compilation.<commit_after>(ns uswitch.bifrost.zk
  (:require [clj-kafka.zk :refer (committed-offset set-offset!)]
            [com.stuartsierra.component :refer (Lifecycle)]
            [clojure.core.async :refer (go <!)]
            [clojure.tools.logging :refer (info error)]))

(defrecord ZookeeperTracker [consumer-properties commit-offset-ch]
  Lifecycle
  (start [this]
    (info "Starting ZookeeperTracker")
    (let [consumer-group-id (consumer-properties "group.id")]
      (go
       (loop [{:keys [topic partition offset] :as m} (<! commit-offset-ch)]
         (when m
           (info "Committing consumer information to ZooKeeper:" m)
           (try
             (set-offset! consumer-properties consumer-group-id topic partition offset)
             (info "Committed offset information to ZooKeeper" m)
             (catch Exception e
               (error e "Unable to commit offset to ZooKeeper")))
           (recur (<! commit-offset-ch))))
       (info "Stopping ZookeeperTracker"))
      this))
  (stop [this]
{'fraction_non_alphanumeric': 0.12415784408084697, 'fraction_numerical': 0.0, 'mean_word_length': 2.603119584055459, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5739694', 'n_tokens_mistral': 640, 'n_tokens_neox': 609, 'n_words': 165}