Skip invalid Kafka messages during Fetch

[ankitk-me]

No description provided.

[jfallows]

Looks good, we just need to improve the log messages.

timestamp namespace.binding: Skipping invalid message on topic abc, partition N, offset M

?

[jfallows]

Later, we're going to be logging to the EngineContext, and we usually don't pass strings around for bindings, just binding ids.

Let's change the method signature to put EngineContext and bindingId as the first 2 parameters. Note that bindingId is already there as routedId in KafkaCacheServerFetchFanout.

Suggested change

	System.out.println(String.format("%s %s: Skipping invalid message on topic %s, partition %d, offset %d",
	System.currentTimeMillis(), bindingName, topic, id, offset));
	System.out.printf("%s:%s %s: Skipping invalid message on topic %s, partition %d, offset %d\n",
	System.currentTimeMillis(), context.supplyNamespace(bindingId), context.supplyLocalName(bindingId), topic, id, offset);

Also, let's put these printfs inside conditional if based on EngineConfiguration.verbose().