KAFKA-19471: Enable acknowledgement for a record which could not be deserialized

[DL1231]

This patch mainly includes two improvements:

1. Update currentFetch when pollForFetches() throws an exception.
2. Add an override KafkaShareConsumer.acknowledge(String topic, int partition, long offset, AcknowledgeType type) .

[AndrewJSchofield]

Thanks for the PR. I've only taken a very cursory look so far and have one comment. I'll review more deeply soon.

[AndrewJSchofield]

If you're providing instances of the deserializers, I don't think you should be setting the class names in the config.

[DL1231]

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

Lines 743 to 744 in 2346c0e

	if (keyDeserializer != null)
	newConfigs.put(KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass());

Thanks for your review. During initialization, if the keyDeserializer is set, it will override the class name in the config.
Of course, the explicit check here makes the logic clearer. I'll fix it later.

[AndrewJSchofield]

Thanks for the PR. I think it's along the right lines, but the validation of the parameters is too loose as it stands. I suggest introducing some kind of exceptional batch that contains a single record, and then validating the parameters against this.

[AndrewJSchofield]

The checking here is not tight enough. If the user calls acknowledge(ConsumerRecord, AcknowledgeType), the code makes sure that the in-flight records include the offset (see ShareInFlightBatch.acknowledge(ConsumerRecord, AcknowledgeType)). However, in this case, there is no validation of the offset. The handling of an exceptional batch needs to be a bit more sophisticated I think. It should not be possible to use ShareConsumer.acknowledge(ConsumerRecord, AcknowledgeType) with an exceptional batch. It should only be possible to use ShareConsumer.acknowlege(String, int, long, AcknowledgeType) and only for the specific parameters that were retrieved from the RecordDeserializationException.