class_docstr: Tests Elastic Horovod on Spark with against a stand-alone Spark cluster.
  
  The cluster is brought up and controlled by the unit test.
  Tests can run forked but not in parallel.
  Running tests in parallel would put CPU under pressure.
  
  Running tests forked will reduce memory footprint from 1.5GB for the master to 512MB.
  Each executor (slot) needs 512GB memory. There are at most 6 executors used in these tests.
  Set SPARK_DRIVER_MEM=512m env when run forked to reduce driver memory usage.
class_docstr_tok: ["Tests","Elastic","Horovod","on","Spark","with","against","a","stand-alone","Spark","cluster",".","The","cluster","is","brought","up","and","controlled","by","the","unit","test",".","Tests","can","run","forked","but","not","in","parallel",".","Running","tests","in","parallel","would","put","CPU","under","pressure",".","Running","tests","forked","will","reduce","memory","footprint","from","1.5GB","for","the","master","to","512MB",".","Each","executor","(","slot",")","needs","512GB","memory",".","There","are","at","most","6","executors","used","in","these","tests",".","Set","SPARK_DRIVER_MEM=512m","env","when","run","forked","to","reduce","driver","memory","usage","."]
code: def test_fault_tolerance_hosts_added_and_removed(self):
      discovery_schedule = [
          (0, ["host-1:1", "host-2:1"]),
          (1, ["host-2:1", "host-3:1"]),
          (None, ["host-3:1", "host-4:1"]),
      ]
  
      # don't wait for discovery of new hosts but have epochs be long enough to see hosts changes
      results = self._run(
          discovery_schedule=discovery_schedule,
          discovery_wait=0,
          epoch_wait=10,
          num_proc=2,
          extra_conf=[
              conf.SPARK_CONF_ALWAYS_RESTART_FAILED_TASK,
              conf.SPARK_CONF_BLACKLIST_DISABLED,
          ],
      )
  
      self.assertEqual(3, len(results))
  
      self.assertEqual(0, results[0]["start_rank"])
      self.assertEqual(2, results[0]["size"])
      self.assertEqual(1, results[0]["rendezvous"])
  
      # either host with rank 0 got removed, or host with rank 1
      if results[1]["start_rank"] == results[0]["start_rank"]:
          #                start rank / rank:
          # epoch    host-1  host-2  host-3  host-4
          #   0       0 / 0   1 / 1
          #   1       0 / 0           1 / 1
          #   2                       1 / 0   1 / 1
  
          # host with start rank 1 got removed in epoch 1, rank 0 unchanged
          self.assertEqual(0, results[1]["start_rank"])
          self.assertEqual(2, results[1]["size"])
          self.assertEqual(
              results[0]["hostname"], results[1]["hostname"]
          )
          self.assertEqual(2, results[1]["rendezvous"])
  
          # host with start rank 0 got removed in epoch 2
          self.assertEqual(1, results[2]["start_rank"])
          self.assertEqual(2, results[2]["size"])
          self.assertNotEqual(
              results[1]["hostname"], results[2]["hostname"]
          )
          self.assertEqual(3, results[2]["rendezvous"])
      else:
          #                start rank / rank:
          # epoch    host-1  host-2  host-3  host-4
          #   0       0 / 0   1 / 1
          #   1               1 / 0   1 / 1
          #   2                       1 / 0   1 / 1
  
          # host with start rank 0 got removed in epoch 1
          self.assertEqual(1, results[1]["start_rank"])
          self.assertEqual(2, results[1]["size"])
          self.assertNotEqual(
              results[0]["hostname"], results[1]["hostname"]
          )
          self.assertEqual(2, results[1]["rendezvous"])
  
          # second host with start rank 1 became rank 0
          self.assertEqual(1, results[2]["start_rank"])
          self.assertEqual(2, results[2]["size"])
          self.assertNotEqual(
              results[0]["hostname"], results[2]["hostname"]
          )
          self.assertNotEqual(
              results[1]["hostname"], results[2]["hostname"]
          )
          self.assertEqual(3, results[2]["rendezvous"])
  
imports: from __future__ import absolute_import
  import contextlib
  import json
  import logging
  import os
  import re
  import sys
  import threading
  import time
  import warnings
  import mock
  from parameterized import parameterized
  import pytest
  import unittest
  from horovod.runner.util.threads import in_thread
  from horovod.runner.common.util import safe_shell_exec, tiny_shell_exec
  from horovod.spark import conf, run_elastic
  from common import temppath
label_desc: Use image node_id 5 for calling the BaseElasticSparkTests obj's underlying member method code with example usage: obj.test_fault_tolerance_hosts_added_and_removed() without return types