Handle/fix "Got unexpected Kafka message" error

[IgorFedchenko]

During test execution under Ubuntu:
Test:
Akka.Streams.Kafka.Tests.Integration.SourceWithOffsetContextIntegrationTests.SourceWithOffsetContext_at_least_once_consuming_should_work

Error:

Failed: Expected a message of type Akka.Streams.TestKit.TestSubscriber+OnNext1[System.ValueTuple2[Akka.NotUsed,Akka.Streams.Kafka.Messages.ICommittableOffsetBatch]], but received {TestSubscriber.OnError(Got unexpected Kafka message: {"Topic":"topic-1-3aac2d51-7efe-4e0c-adca-0bac512289e9","Partition":{"Value":0,"IsSpecial":false},"Offset":{"Value":0,"IsSpecial":false},"TopicPartition":{"Topic":"topic-1-3aac2d51-7efe-4e0c-adca-0bac512289e9","Partition":{"Value":0,"IsSpecial":false}},"TopicPartitionOffset":{"Topic":"topic-1-3aac2d51-7efe-4e0c-adca-0bac512289e9","Partition":{"Value":0,"IsSpecial":false},"Offset":{"Value":0,"IsSpecial":false},"TopicPartition":{"Topic":"topic-1-3aac2d51-7efe-4e0c-adca-0bac512289e9","Partition":{"Value":0,"IsSpecial":false}}},"Message":{"Value":"3","Timestamp":{"Type":1,"UnixTimestampMs":1580383308079,"UtcDateTime":"2020-01-30T11:21:48.079Z"},"Headers":[]},"Value":"3","Timestamp":{"Type":1,"UnixTimestampMs":1580383308079,"UtcDateTime":"2020-01-30T11:21:48.079Z"},"Headers":[],"IsPartitionEOF":false})} (type Akka.Streams.TestKit.TestSubscriber+OnError) instead from [akka://SourceWithOffsetContextIntegrationTests/user/StreamSupervisor-27/Flow-1-0-unknown-operation#1937253766]

Expected: True

Actual: False

Details: https://dev.azure.com/dotnet/Akka.NET/_build/results?buildId=27932&view=ms.vss-test-web.build-test-results-tab&runId=1085592&resultId=100006&paneView=debug

It is thrown from KafkaConsumerActor:

Akka.Streams.Kafka/src/Akka.Streams.Kafka/Stages/Consumers/Actors/KafkaConsumerActor.cs

Lines 304 to 311 in 875b806

	// no outstanding requests so we don't expect any messages back, but we should anyway
	// drive the KafkaConsumer by polling to handle partition events etc.
	PausePartitions(currentAssignment);
	var message = _consumer.Consume(TimeSpan.FromMilliseconds(1));
	if (message != null)
	{
	throw new IllegalActorStateException($"Got unexpected Kafka message: {message.ToJson()}");
	}

Basically partition should be paused, but for some reason message is still received.

I think that instead of throwning exception here, we can log warning and make a workaround - ignore message and reset partition position pointer by 1 .