[Failed] CommittingWithMockSpec should collect commits to be sent to commitAsync

[ennru]

[info] CommittingWithMockSpec:

[info] - should commit metadata in message (469 milliseconds)

[info] - should call commitAsync for commit message and then complete future (228 milliseconds)

[info] - should fail future in case of commit fail (220 milliseconds)

[info] - should collect commits to be sent to commitAsync *** FAILED *** (10 seconds, 101 milliseconds)

[info]   java.util.concurrent.TimeoutException: Futures timed out after [10 seconds]

[info]   at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:259)

[info]   at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:263)

[info]   at scala.concurrent.Await$.$anonfun$result$1(package.scala:220)

[info]   at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:57)

[info]   at scala.concurrent.Await$.result(package.scala:146)

[info]   at akka.kafka.internal.CommittingWithMockSpec.$anonfun$new$13(CommittingWithMockSpec.scala:229)

https://travis-ci.org/akka/alpakka-kafka/jobs/581091376#L827

[seglo]

https://travis-ci.org/akka/alpakka-kafka/jobs/592616580#L832

[seglo]

https://travis-ci.org/akka/alpakka-kafka/jobs/595824299#L591

[seglo]

https://travis-ci.org/akka/alpakka-kafka/jobs/640314577#L650

[ennru]

https://travis-ci.org/akka/alpakka-kafka/jobs/640635808#L649 in #1026

[ennru]

https://travis-ci.org/akka/alpakka-kafka/jobs/648052162#L586 in master

[ignasi35]

I think the problem is on :

alpakka-kafka/tests/src/test/scala/akka/kafka/internal/CommittingWithMockSpec.scala

Lines 216 to 222 in 4f7760f

	withClue("the commits are aggregated to a low number of calls to commitAsync:") {
	awaitAssert {
	val callsToCommitAsync = commitLog.calls.size
	callsToCommitAsync should be >= 1
	callsToCommitAsync should be < count / 10
	}
	}

It's a block that attempts to wait until all the commits have been submitted by counting how many callbacks were registered. The invocations to commitAsync may group messages so that block doesn't await until 100 callbacks are registered, and, instead, blocks until a few have been registered. In slow VM's this barrier is open when 1 or 2 callbacks have been registered but a lot more are still missing.

The consequence is that emulating the commit:

alpakka-kafka/tests/src/test/scala/akka/kafka/internal/CommittingWithMockSpec.scala

Lines 224 to 230 in 4f7760f

	//emulate commit
	commitLog.calls.foreach {
	case (offsets, callback) => callback.onComplete(offsets.asJava, null)
	}
	allCommits.futureValue should have size (count.toLong)
	control.shutdown().futureValue shouldBe Done
	}

then is not invoking all the callbacks but just the first few blocks (the fast ones).

This leaves a handful of messages uncommited causing the test to fail.

This can be reproduced locally commenting the await block:

//    withClue("the commits are aggregated to a low number of calls to commitAsync:") {
//      awaitAssert {
//        val callsToCommitAsync = commitLog.calls.size
//        callsToCommitAsync should be >= 1
//        callsToCommitAsync should be < count / 10
//      }
//    }

[ignasi35]

Fixed (?) in #1047. We can reopen if the issue is not solved completely.

[ennru]

It shows in a new variant in master:

- should collect commits to be sent to commitAsync *** FAILED *** (1 minute, 40 seconds)
  the commits are aggregated to a low number of calls to commitAsync: 11 was not less than 10 (CommittingWithMockSpec.scala:223)
  org.scalatest.exceptions.TestFailedException:
  at org.scalatest.MatchersHelper$.indicateFailure(MatchersHelper.scala:343)
  at org.scalatest.Matchers$ShouldMethodHelper$.shouldMatcher(Matchers.scala:6723)
  at org.scalatest.Matchers$AnyShouldWrapper.should(Matchers.scala:6759)
  at akka.kafka.internal.CommittingWithMockSpec.$anonfun$new$18(CommittingWithMockSpec.scala:223)

https://travis-ci.org/github/akka/alpakka-kafka/jobs/674754524#L660

[seglo]

It's back.

https://travis-ci.org/github/akka/alpakka-kafka/jobs/696759317#L626

--> [akka.kafka.internal.CommittingWithMockSpec: should collect commits to be sent to commitAsync] Start of log messages of test that [Failed(org.scalatest.exceptions.TestFailedException: the commits are aggregated to a low number of calls to commitAsync: 13 was not less than 10)]

[seglo]

https://travis-ci.org/github/akka/alpakka-kafka/jobs/718740561#L810