<commit_before>(ns backtype.storm.zookeeper
  (:import [org.apache.zookeeper ZooKeeper Watcher KeeperException$NoNodeException
            ZooDefs ZooDefs$Ids CreateMode WatchedEvent Watcher$Event Watcher$Event$KeeperState
            Watcher$Event$EventType])
  (:import [org.apache.zookeeper.data Stat])
  (:import [org.apache.zookeeper.server ZooKeeperServer NIOServerCnxn$Factory])
  (:import [java.net InetSocketAddress])
  (:import [java.io File])
  (:use [backtype.storm util log config]))

(def zk-keeper-states
  {Watcher$Event$KeeperState/Disconnected :disconnected
   Watcher$Event$KeeperState/SyncConnected :connected
   Watcher$Event$KeeperState/AuthFailed :auth-failed
   Watcher$Event$KeeperState/Expired :expired
  })

(def zk-event-types
  {Watcher$Event$EventType/None :none
   Watcher$Event$EventType/NodeCreated :node-created
   Watcher$Event$EventType/NodeDeleted :node-deleted
   Watcher$Event$EventType/NodeDataChanged :node-data-changed
   Watcher$Event$EventType/NodeChildrenChanged :node-children-changed
  })


;; TODO: make this block until session is established (wait until a flag is triggered by watcher)
(defn mk-client
  ([conn-str session-timeout watcher]
    (ZooKeeper.
      conn-str
      session-timeout
      (reify Watcher
        (^void process [this ^WatchedEvent event]
          (watcher (zk-keeper-states (.getState event))
                   (zk-event-types (.getType event))
                   (.getPath event))
          ))))
  ([conn-str watcher]
    (mk-client conn-str 10000 watcher))
  ([conn-str]
    ;; this constructor is intended for debugging
    (mk-client
      conn-str
      (fn [state type path]
        (log-message "Zookeeper state update: " state type path)))
      ))

(def zk-create-modes
  {:ephemeral CreateMode/EPHEMERAL
   :persistent CreateMode/PERSISTENT})

(defn create-node
  ([^ZooKeeper zk ^String path ^bytes data mode]
    (.create zk (normalize-path path) data ZooDefs$Ids/OPEN_ACL_UNSAFE (zk-create-modes mode)))
  ([^ZooKeeper zk ^String path ^bytes data]
    (create-node zk path data :persistent)))

(defn exists-node? [^ZooKeeper zk ^String path watch?]
  ((complement nil?) (.exists zk (normalize-path path) watch?)))

(defn delete-node [^ZooKeeper zk ^String path]
  (.delete zk (normalize-path path) -1))

(defn mkdirs [^ZooKeeper zk ^String path]
  (let [path (normalize-path path)]
    (when-not (or (= path "/") (exists-node? zk path false))
      (mkdirs zk (parent-path path))
      (create-node zk path (barr 7) :persistent)
      )))

(defn get-data [^ZooKeeper zk ^String path watch?]
  (let [path (normalize-path path)]
    (try
      (if (.exists zk path watch?)
        (.getData zk path watch? (Stat.)))
    (catch KeeperException$NoNodeException e
      ;; this is fine b/c we still have a watch from the successful exists call
      nil ))))
<commit_msg>fix race condition in zookeeper#mkdirs<commit_after>(ns backtype.storm.zookeeper
  (:import [org.apache.zookeeper ZooKeeper Watcher KeeperException$NoNodeException
            ZooDefs ZooDefs$Ids CreateMode WatchedEvent Watcher$Event Watcher$Event$KeeperState
            Watcher$Event$EventType KeeperException$NodeExistsException])
  (:import [org.apache.zookeeper.data Stat])
  (:import [org.apache.zookeeper.server ZooKeeperServer NIOServerCnxn$Factory])
  (:import [java.net InetSocketAddress])
  (:import [java.io File])
  (:use [backtype.storm util log config]))

(def zk-keeper-states
  {Watcher$Event$KeeperState/Disconnected :disconnected
   Watcher$Event$KeeperState/SyncConnected :connected
   Watcher$Event$KeeperState/AuthFailed :auth-failed
   Watcher$Event$KeeperState/Expired :expired
  })

(def zk-event-types
  {Watcher$Event$EventType/None :none
   Watcher$Event$EventType/NodeCreated :node-created
   Watcher$Event$EventType/NodeDeleted :node-deleted
   Watcher$Event$EventType/NodeDataChanged :node-data-changed
   Watcher$Event$EventType/NodeChildrenChanged :node-children-changed
  })


;; TODO: make this block until session is established (wait until a flag is triggered by watcher)
(defn mk-client
  ([conn-str session-timeout watcher]
    (ZooKeeper.
      conn-str
      session-timeout
      (reify Watcher
        (^void process [this ^WatchedEvent event]
          (watcher (zk-keeper-states (.getState event))
                   (zk-event-types (.getType event))
                   (.getPath event))
          ))))
  ([conn-str watcher]
    (mk-client conn-str 10000 watcher))
  ([conn-str]
    ;; this constructor is intended for debugging
    (mk-client
      conn-str
      (fn [state type path]
        (log-message "Zookeeper state update: " state type path)))
      ))

(def zk-create-modes
  {:ephemeral CreateMode/EPHEMERAL
   :persistent CreateMode/PERSISTENT})

(defn create-node
  ([^ZooKeeper zk ^String path ^bytes data mode]
    (.create zk (normalize-path path) data ZooDefs$Ids/OPEN_ACL_UNSAFE (zk-create-modes mode)))
  ([^ZooKeeper zk ^String path ^bytes data]
    (create-node zk path data :persistent)))

(defn exists-node? [^ZooKeeper zk ^String path watch?]
  ((complement nil?) (.exists zk (normalize-path path) watch?)))

(defn delete-node [^ZooKeeper zk ^String path]
  (.delete zk (normalize-path path) -1))

(defn mkdirs [^ZooKeeper zk ^String path]
  (let [path (normalize-path path)]
    (when-not (or (= path "/") (exists-node? zk path false))
      (mkdirs zk (parent-path path))
      (try
        (create-node zk path (barr 7) :persistent)
        (catch KeeperException$NodeExistsException e
          ;; this can happen when multiple clients doing mkdir at same time
          ))
      )))

(defn get-data [^ZooKeeper zk ^String path watch?]
  (let [path (normalize-path path)]
    (try
      (if (.exists zk path watch?)
        (.getData zk path watch? (Stat.)))
    (catch KeeperException$NoNodeException e
      ;; this is fine b/c we still have a watch from the successful exists call
      nil ))))

{'fraction_non_alphanumeric': 0.11618887015177065, 'fraction_numerical': 0.002360876897133221, 'mean_word_length': 3.503416856492027, 'pattern_counts': {'":': 0, '<': 3, '<?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': '5378855', 'n_tokens_mistral': 2033, 'n_tokens_neox': 1914, 'n_words': 512}