Fix #252 kafka sink user error

[breezykermo]

Closes #252 by communicating the error to the controller (following the pattern in the Kafka source).

When running in development with a basic job that plugs data from the nexmark source to a (badly configured) kafka sink, i.e.:

insert into badkaf
select process_string(person.name) from nex where person.name is not null;

(where process_string simply ensures a String is always returned, rather than an Option<String> -- as my badkaf sink expects a raw string.)

Arroyo seems to immediately attempt to go into a checkpoint, and eventually (after several minutes) times out through a failed self.flush. When this happens, the error is appropriately displayed in the UX in my setup:

I am not familiar enough with the lifecycle to understand how to test the error code path in self.process_element given that the setup described seems to immediately go into a checkmark. I would appreciate guidance on how to do this, and will be happy to update the PR accordingly.

I haven't added a test to catch this error/panic specifically, as from what I can tell the testing in connectors only confirms happy path.

[mwylde]

Thanks for the contribution, this looks great!