Pass executor into stream router engine.

Avoids metrics leak with instrumented executors.
Improve exception handling on timeouts.
Add proper name for InstrumentedExecutorService.
Add getFieldNames() to Message and use it instead of getFields().

This avoids pulling an immutable copy of the map out of the message when
it is not needed.
Add hasField() to Message.
Use message.hasField() to avoid expensive getFields();
Cache getId() in PersistedImpl to avoid expensive toHexString() calls.