Allow StreamConsumer to reject partition from KafkaIngester

[cpwright]

As an integrator, I would like more control over the StreamConsumers that are created from the KafkaIngester. In particular, I may find that some partitions cannot be ingested by my backend.

If I throw an Exception, then the entire ingestion process is going to be terminated. If I return a do-nothing ingester, then there will still be additional effort expended processing those partitions.

Allowing the Function<TopicPartition, KafkaRecordConsumer> to return null and behave similarly to filtering out the partition, would provide backwards compatible behavior and more flexibility.