docs: public static void main(String[] args) throws Exception {
  
  		// ---- print some usage help ----
  
  		System.out.println("Usage with built-in data generator: StateMachineExample [--error-rate <probability-of-invalid-transition>] [--sleep <sleep-per-record-in-ms>]");
  		System.out.println("Usage with Kafka: StateMachineExample --kafka-topic <topic> [--brokers <brokers>]");
  		System.out.println("Options for both the above setups: ");
  		System.out.println("\t[--backend <file|rocks>]");
  		System.out.println("\t[--checkpoint-dir <filepath>]");
  		System.out.println("\t[--async-checkpoints <true|false>]");
  		System.out.println("\t[--incremental-checkpoints <true|false>]");
  		System.out.println("\t[--output <filepath> OR null for stdout]");
  		System.out.println();
  
  		// ---- determine whether to use the built-in source, or read from Kafka ----
  
  		final SourceFunction<Event> source;
  		final ParameterTool params = ParameterTool.fromArgs(args);
  
  		if (params.has("kafka-topic")) {
  			// set up the Kafka reader
  			String kafkaTopic = params.get("kafka-topic");
  			String brokers = params.get("brokers", "localhost:9092");
  
  			System.out.printf("Reading from kafka topic %s @ %s\n", kafkaTopic, brokers);
  			System.out.println();
  
  			Properties kafkaProps = new Properties();
  			kafkaProps.setProperty("bootstrap.servers", brokers);
  
  			FlinkKafkaConsumer010<Event> kafka = new FlinkKafkaConsumer010<>(kafkaTopic, new EventDeSerializer(), kafkaProps);
  			kafka.setStartFromLatest();
  			kafka.setCommitOffsetsOnCheckpoints(false);
  			source = kafka;
  		}
  		else {
  			double errorRate = params.getDouble("error-rate", 0.0);
  			int sleep = params.getInt("sleep", 1);
  
  			System.out.printf("Using standalone source with error rate %f and sleep delay %s millis\n", errorRate, sleep);
  			System.out.println();
  
  			source = new EventsGeneratorSource(errorRate, sleep);
  		}
  
  		// ---- main program ----
  
  		// create the environment to create streams and configure execution
  		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  		env.enableCheckpointing(2000L);
  
  		final String stateBackend = params.get("backend", "memory");
  		if ("file".equals(stateBackend)) {
  			final String checkpointDir = params.get("checkpoint-dir");
  			boolean asyncCheckpoints = params.getBoolean("async-checkpoints", false);
  			env.setStateBackend(new FsStateBackend(checkpointDir, asyncCheckpoints));
  		} else if ("rocks".equals(stateBackend)) {
  			final String checkpointDir = params.get("checkpoint-dir");
  			boolean incrementalCheckpoints = params.getBoolean("incremental-checkpoints", false);
  			env.setStateBackend(new RocksDBStateBackend(checkpointDir, incrementalCheckpoints));
  		}
  
  		final String outputFile = params.get("output");
  
  		// make parameters available in the web interface
  		env.getConfig().setGlobalJobParameters(params);
  
  		DataStream<Event> events = env.addSource(source);
  
  		DataStream<Alert> alerts = events
  				// partition on the address to make sure equal addresses
  				// end up in the same state machine flatMap function
  				.keyBy(Event::sourceAddress)
  
  				// the function that evaluates the state machine over the sequence of events
  				.flatMap(new StateMachineMapper());
  
  		// output the alerts to std-out
  		if (outputFile == null) {
  			alerts.print();
  		} else {
  			alerts
  				.writeAsText(outputFile, FileSystem.WriteMode.OVERWRITE)
  				.setParallelism(1);
  		}
  
  		// trigger program execution
  		env.execute("State machine job");
  	}