Feature Request: Setting up a Highly Available (HA) pipeline with Logstash Nodes

As a kibana user watching transactional data, I would like to configure a logstash pipeline with strong guarantees that events are not lost so that I can use ELK as a forensic analysis tool.

In logstash 1.4, the pipeline currently reads off a connection and acks as soon as it's made it into memory.

This is a problem if the underlying software or the hardware were to fail while logstash is processing a message. Some examples of software failure:
- Operating system stability
- Codec stacktrace on edge case (e.g. network outages on virtualised switches)
- VM Hypervisor failure

Scope:
- This change will extend 'the life of an event in logstash'
- Extend the lumberjack Input with needs_ha option
- Extend the elasticsearch Output with provides_ha option
- Opt in: both needs_ha and provides_ha should be opt in as it will likely have a performance hit

Out of scope:
- Existing Inputs and Outputs plugins should still operate without code changes
- Setting up clustering technologies to be Highly Available (HA)

Requirements:
- The Pipeline should only allow one HA output to simplify when an 'ack' is ready to be sent
- The Pipeline should allow one or more inputs to ack on delivery to an HA output
- Outputs labeled provides_ha should only write when all events have been processed from the input's spooled batch
- Input labeled needs_ha should only ack spooled event batches when all events have been written to the HA Output
- Clients who have not received an ack should resend if disconnect happens before they're acked. Most clients which implement ack already do this.
- Events should be cancelled as soon as possible when an Input disconnects to stop them writing to an Output.
- Cancelled events from drop{} should not block other events being written/acked.
- Cloned events should only ack on delivery of both messages to an HA output.

Configuration example:

```
input {
lumberjack {
port => 2000
ssl_certificate => "/path/to/selfsigned.crt"
ssl_key => "/path/to/selfsigned.key"
needs_ha => true
}
}

output {
elasticsearch {
host => "172.17.0.2"
provides_ha => true
}
}
```