<commit_before>(ns onyx.windowing.basic-windowing-crash-test
  (:require [clojure.core.async :refer [chan >!! <!! close! sliding-buffer]]
            [clojure.test :refer [deftest is]]
            [taoensso.timbre :refer [info error warn trace fatal] :as timbre]
            [onyx.tasks.seq]
            [onyx.plugin.core-async :refer [take-segments!]]
            [onyx.static.uuid :refer [random-uuid]]
            [onyx.state.state-extensions :as state-extensions]
            [onyx.test-helper :refer [load-config with-test-env add-test-env-peers!]]
            [onyx.job :refer [add-task]]
            [onyx.api]))

(def input
  [{:id 1  :age 21 :event-time #inst "2015-09-13T03:00:00.829-00:00"}
   {:id 2  :age 12 :event-time #inst "2015-09-13T03:04:00.829-00:00"}
   {:id 3  :age 3  :event-time #inst "2015-09-13T03:05:00.829-00:00"}
   {:id 4  :age 64 :event-time #inst "2015-09-13T03:06:00.829-00:00"}
   {:id 5  :age 53 :event-time #inst "2015-09-13T03:07:00.829-00:00"}
   {:id 6  :age 52 :event-time #inst "2015-09-13T03:08:00.829-00:00"}
   {:id 7  :age 24 :event-time #inst "2015-09-13T03:09:00.829-00:00"}
   {:id 8  :age 35 :event-time #inst "2015-09-13T03:15:00.829-00:00"}
   {:id 9  :age 49 :event-time #inst "2015-09-13T03:25:00.829-00:00"}
   {:id 10 :age 37 :event-time #inst "2015-09-13T03:45:00.829-00:00"}
   {:id 11 :age 15 :event-time #inst "2015-09-13T03:03:00.829-00:00"}
   {:id 12 :age 22 :event-time #inst "2015-09-13T03:56:00.829-00:00"}
   {:id 13 :age 83 :event-time #inst "2015-09-13T03:59:00.829-00:00"}
   {:id 14 :age 60 :event-time #inst "2015-09-13T03:32:00.829-00:00"}
   {:id 15 :age 35 :event-time #inst "2015-09-13T03:16:00.829-00:00"}
   ;; Ensure some duplicate ages are counted, with different ids
   {:id 16  :age 12 :event-time #inst "2015-09-13T03:04:00.829-00:00"}
   {:id 17  :age 52 :event-time #inst "2015-09-13T03:08:00.829-00:00"}
   {:id 18  :age 53 :event-time #inst "2015-09-13T03:07:00.829-00:00"}
   {:id 19  :age 3 :event-time #inst "2015-09-13T03:05:00.829-00:00"}])

(defn output->final-counts [window-counts]
  (let [grouped (group-by (juxt first second) window-counts)]
    (set (map (fn get-latest [[[bounds k] all]]
                [bounds k (last (sort (map last all)))])
              grouped)))) 

(def expected-windows
  #{[[1442115000000 1442115299999] 60 1]
    [[1442113200000 1442113499999] 21 1]
    [[1442113500000 1442113799999] 64 1]
    [[1442113500000 1442113799999] 3 2]
    [[1442116500000 1442116799999] 83 1]
    [[1442115900000 1442116199999] 37 1]
    [[1442113500000 1442113799999] 52 2]
    [[1442114100000 1442114399999] 35 2]
    [[1442113500000 1442113799999] 53 2]
    [[1442113200000 1442113499999] 15 1]
    [[1442114700000 1442114999999] 49 1]
    [[1442113500000 1442113799999] 24 1]
    [[1442113200000 1442113499999] 12 2]
    [[1442116500000 1442116799999] 22 1]})

(defn restartable? [event lifecycle lifecycle-name e]
  :restart)

(def test-state (atom nil))

(def batch-num (atom nil))

(def in-chan (atom nil))
(def in-buffer (atom nil))

(def out-chan (atom nil))

(defn update-atom! [event window trigger {:keys [lower-bound upper-bound group-key event-type] :as opts} extent-state]
  (info "TRIGGERING" group-key lower-bound upper-bound extent-state)
  (when-not (= :job-completed event-type)
    (swap! test-state conj [[lower-bound upper-bound] group-key extent-state])))

(def crashed? (atom false))

(def identity-crash
  {:lifecycle/handle-exception restartable?
   :lifecycle/after-read-batch (fn [event lifecycle]
                                 (when (or (and (some #(= (:id %) 16)
                                                      (:onyx.core/batch event))
                                                (not @crashed?))
                                           (zero? (rand-int 100)))
                                   (println "CRASH IT")
                                   (reset! crashed? true)
                                  (throw (ex-info "Restart me!" {})))
                                 {})})

(defn inject-in-ch [event lifecycle]
  {:core.async/chan @in-chan})

(defn inject-out-ch [event lifecycle]
  {:core.async/chan @out-chan})

(def in-calls
  {:lifecycle/before-task-start inject-in-ch})

(def out-calls
  {:lifecycle/before-task-start inject-out-ch})

(defn at-out [segment]
  (info "AT OUT SEG" segment)
  segment)

(deftest fault-tolerance-fixed-windows-segment-trigger
  (do<commit_msg>Reduce number of windowing crashes<commit_after>(ns onyx.windowing.basic-windowing-crash-test
  (:require [clojure.core.async :refer [chan >!! <!! close! sliding-buffer]]
            [clojure.test :refer [deftest is]]
            [taoensso.timbre :refer [info error warn trace fatal] :as timbre]
            [onyx.tasks.seq]
            [onyx.plugin.core-async :refer [take-segments!]]
            [onyx.static.uuid :refer [random-uuid]]
            [onyx.state.state-extensions :as state-extensions]
            [onyx.test-helper :refer [load-config with-test-env add-test-env-peers!]]
            [onyx.job :refer [add-task]]
            [onyx.api]))

(def input
  [{:id 1  :age 21 :event-time #inst "2015-09-13T03:00:00.829-00:00"}
   {:id 2  :age 12 :event-time #inst "2015-09-13T03:04:00.829-00:00"}
   {:id 3  :age 3  :event-time #inst "2015-09-13T03:05:00.829-00:00"}
   {:id 4  :age 64 :event-time #inst "2015-09-13T03:06:00.829-00:00"}
   {:id 5  :age 53 :event-time #inst "2015-09-13T03:07:00.829-00:00"}
   {:id 6  :age 52 :event-time #inst "2015-09-13T03:08:00.829-00:00"}
   {:id 7  :age 24 :event-time #inst "2015-09-13T03:09:00.829-00:00"}
   {:id 8  :age 35 :event-time #inst "2015-09-13T03:15:00.829-00:00"}
   {:id 9  :age 49 :event-time #inst "2015-09-13T03:25:00.829-00:00"}
   {:id 10 :age 37 :event-time #inst "2015-09-13T03:45:00.829-00:00"}
   {:id 11 :age 15 :event-time #inst "2015-09-13T03:03:00.829-00:00"}
   {:id 12 :age 22 :event-time #inst "2015-09-13T03:56:00.829-00:00"}
   {:id 13 :age 83 :event-time #inst "2015-09-13T03:59:00.829-00:00"}
   {:id 14 :age 60 :event-time #inst "2015-09-13T03:32:00.829-00:00"}
   {:id 15 :age 35 :event-time #inst "2015-09-13T03:16:00.829-00:00"}
   ;; Ensure some duplicate ages are counted, with different ids
   {:id 16  :age 12 :event-time #inst "2015-09-13T03:04:00.829-00:00"}
   {:id 17  :age 52 :event-time #inst "2015-09-13T03:08:00.829-00:00"}
   {:id 18  :age 53 :event-time #inst "2015-09-13T03:07:00.829-00:00"}
   {:id 19  :age 3 :event-time #inst "2015-09-13T03:05:00.829-00:00"}])

(defn output->final-counts [window-counts]
  (let [grouped (group-by (juxt first second) window-counts)]
    (set (map (fn get-latest [[[bounds k] all]]
                [bounds k (last (sort (map last all)))])
              grouped)))) 

(def expected-windows
  #{[[1442115000000 1442115299999] 60 1]
    [[1442113200000 1442113499999] 21 1]
    [[1442113500000 1442113799999] 64 1]
    [[1442113500000 1442113799999] 3 2]
    [[1442116500000 1442116799999] 83 1]
    [[1442115900000 1442116199999] 37 1]
    [[1442113500000 1442113799999] 52 2]
    [[1442114100000 1442114399999] 35 2]
    [[1442113500000 1442113799999] 53 2]
    [[1442113200000 1442113499999] 15 1]
    [[1442114700000 1442114999999] 49 1]
    [[1442113500000 1442113799999] 24 1]
    [[1442113200000 1442113499999] 12 2]
    [[1442116500000 1442116799999] 22 1]})

(defn restartable? [event lifecycle lifecycle-name e]
  :restart)

(def test-state (atom nil))

(def batch-num (atom nil))

(def in-chan (atom nil))
(def in-buffer (atom nil))

(def out-chan (atom nil))

(defn update-atom! [event window trigger {:keys [lower-bound upper-bound group-key event-type] :as opts} extent-state]
  (info "TRIGGERING" group-key lower-bound upper-bound extent-state)
  (when-not (= :job-completed event-type)
    (swap! test-state conj [[lower-bound upper-bound] group-key extent-state])))

(def identity-crash
  {:lifecycle/handle-exception restartable?
   :lifecycle/after-read-batch (fn [event lifecycle]
                                 (when (and (not (empty? (:onyx.core/batch event))) 
                                                (zero? (rand-int 6)))
                                  (throw (ex-info "Restart me!" {})))
                                 {})})

(defn inject-in-ch [event lifecycle]
  {:core.async/chan @in-chan})

(defn inject-out-ch [event lifecycle]
  {:core.async/chan @out-chan})

(def in-calls
  {:lifecycle/before-task-start inject-in-ch})

(def out-calls
  {:lifecycle/before-task-start inject-out-ch})

(defn at-out [segment]
  (info "AT OUT SEG" segment)
  segment)

(deftest fault-tolerance-fixed-windows-segment-trigger
  (do
{'fraction_non_alphanumeric': 0.1642915073572008, 'fraction_numerical': 0.20206233344919478, 'mean_word_length': 2.782646801051709, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1686122', 'n_tokens_mistral': 4499, 'n_tokens_neox': 3401, 'n_words': 769}