KAFKA-9748: Add Streams eos-beta integration test

[mjsax]

Call for review @abbccdda @guozhangwang

[abbccdda]

Thanks for the added test, left a couple of comments. One high level question I have is what's the expected behavior under fault injection? It seems we skipped a lot of validations along the way.

[abbccdda]

nit: we could define this transition list in a variable to be reused.

[abbccdda]

What about having separate wait condition, or at least have a way to detect which instance gets stuck?

[abbccdda]

Could we define 10 as a constant COMMIT_MARKER value, so that people understand that it has a special meaning to indicate a commit request?

[mjsax]

Not sure if this buys much? The test is complex and one needs to read the comments anyway. We also have a commit marker after 20 and 30 messages and I think using

prepareData(15L, 2 * COMMIT_MARKER, 0L, 1L);

does not improve readablility?

[abbccdda]

Could we just use a boolean flag as parameter to determine whether to only read committed data?

[guozhangwang]

+1 It seems we do not need the actual groupId here, just a boolean flag.

[abbccdda]

Could we pass a Collections.emptyMap() here instead?

[mjsax]

That does not work, because computeExpectedResult will modify the passed in HashMap but Collections.emptyMap() returns an immutable map.

[abbccdda]

nit: s/state/states

[abbccdda]

nit: extra space after second COMMIT

[guozhangwang]

Made a pass.

[guozhangwang]

Why the second client will have two pending transactions? Upon migrated the task the initTxn should cause the pending transaction failed.

[mjsax]

The second client is the "healthy" one: it will host one task and has one open transaction before the crash. The first client also hosts one task before the crash. When the first client fails, it does not have a pending TX because it did not write any output records yet (thus there is nothing to be aborted) and its task will migrate to the second client. The second client won't "touch" its original task and only create a second task, start a new TX for the new task (it would call init and would abort a pending TX if there would be any, but is also starts it's own new TX) and process 5 records (ie, retry the not yet processed record 10 to 14) without committing the TX because it would only commit after 10 records. Hence, it "stabilizes" with 2 pending transactions with 5 writes each.

Does this make sense?

Thinking about it: should we inject the error only after the first client wrote some pending output records to the output topic to verify if the TX gets aborted?

[guozhangwang]

I thought the first client would only crash after it has processed 5 records and hence there's an ongoing transaction already?

[mjsax]

Not in the current way the test is written...

[guozhangwang]

Not sure why it's the case? I think the previous pending txn should have aborted in step 4.

[mjsax]

Good catch. The commit happens for both cases, because when we migrate the task to the restarted client, the "old" client would commit the task. Will update the comment.

[guozhangwang]

In the current settings, with either alpha or beta, we will have one producer per thread since each thread would host one task only, right? Should we have 4 partitions so that under alpha we will have two producers and two txns per thread?

[mjsax]

That is correct. I did consider using 4 partitions for the same reason, but was not sure if it would add value to the test? In the end, the "gap" between eos-alpha and eos-beta is not the number to open transaction, but the usage of different transactional.ids between a task-producer and a thread-producer and this gap is closed via "fetch offset fencing". Hence, if the "fetch offset fencing" works for one task-producer vs one thread-producer (both using different txId), it also works for two task-producers vs one thread-producer?

Thoughts?

[guozhangwang]

What I'm thinking is when there are two running clients, one with eos-alpha and another with eos-beta (upgraded from eos-alpha) the number of transaction.ids is actually reduced, and hence the number of max in-flight txns, and logically I agree they should not have much impact, but hey without the testing we don't know about what we don't know right? If you think such "changes of number of txn.ids and hence number of txns" has been covered in other system tests then probably it's fine. But if using 4 partitions isn't going to make the test more complex / takes much longer, could we be a bit over-cautious here?

[mjsax]

Sure. Works for me.

[guozhangwang]

Should we make this a concurrent linked list since concurrent threads may be adding at the same time?

[mjsax]

There should not be any concurrent threads, because stream1Alpha and streams1Beta don't run at the same time but strictly one after each other.

[guozhangwang]

Ah, you're right.

[guozhangwang]

Seems we can just keep errorInjected to null in phase 1 since we do not need to inject failures?

[mjsax]

This is a little tricky... We get a handle on the AtomicBoolean within the Processor#init() method. In phase 1 we create the Processor and thus we need to prepare the error injection at this point already. In the later phases, Processor#init() is not called for the processors hosted by client-1 any more.

Does this make sense?

[guozhangwang]

Got it, makes sense.

[guozhangwang]

+1 It seems we do not need the actual groupId here, just a boolean flag.

[guozhangwang]

After it resumes to running, p-0 should be aborted right?

[mjsax]

Well, there is no TX for p-0 on client-1 as it crashes before it writes a first output record and thus never starts a TX. (Compare my comment from above.)

[guozhangwang]

Here p-0 may end with COMMIT (close) or ABORT (crash) right?

[mjsax]

Same comment as above: the is no pending TX for p-0 when a crash happens. But the comment here is still simplfied. The end state would be (in full details):

            // phase 5:
            // expected end state per output partition:
            // stop case (original task migrated back):
            //   p-0: 10 records + COMMIT + 5 records + COMMIT
            //   p-1: 10 records + COMMIT + 5 records (pending)
            // stop case (task "switch"):
            //   p-0: 10 records + COMMIT + 5 records + COMMIT
            //   p-1: 10 records + COMMIT + 5 records + COMMIT
            // crash case (original task migrates back):
            //   p-0: 10 records + COMMIT + 5 records + COMMIT
            //   p-1: 10 records + COMMIT + 5 records (pending)
            // crash case (task "switch"):
            //   p-0: 10 records + COMMIT + 5 records (pending)
            //   p-1: 10 records + COMMIT + 5 records + COMMIT

[guozhangwang]

I thought that before the client crashed there's a pending txn because we've processed 5 records already in stage 3):

3.  write 5 records per input topic partition to get pending transactions (verified via "read_uncommitted" mode)
        //      - both pending transactions are based on task producers

Is that not right?

[mjsax]

In the current implementation, we set the inject error flag to true, before we write the 5 input records. Hence, we crash when processing the first record -- therefore, no output record is written and no TX is started (as we start TX lazily).

We cannot set the inject error flag to true after we sent the input data, because we would be subject to a race condition -- all 5 records might have been processed before we set the flag and thus, so error would happen as process() won't be called any more. We can of course change the test to add an additional condition in the Processor to only throw an exception if the flag is set to true and if, eg, value % 10 == 4 (ie, we throw when processing the last record, and thus get 4 pending writes).

[guozhangwang]

This is a meta comment: I think the key here is to check that the offsets sent via "sendOffsetsToTxn" would not be seen to the new owner of the task, i.e. the fencing would take effects. But here since we do not commit (only 5 out of 10 records) this logic would never be checked.

On the other hand, inside StreamsProducer we always call

            producer.sendOffsetsToTransaction(offsets, consumerGroupMetadata);
            producer.commitTransaction();

together, it is a bit hard to check with the given producer.

I'd suggest doing a bit different here: we use a customized client-supplier that generate an extended producer which, depending on the failure injection boolean, throws an exception in commitTransaction immediately, so that we made sure the sendOffsetsToTransaction completes (i.e. the request acked from group coordinator already) before we crash, instead of crashing in the middle of the processing where we'd not send any offsets to group coordinator anyways.

[mjsax]

Correct. I was aware that this case would not be covered, but did not have a good idea how to test it. Your idea is great! Will work on that.

[mjsax]

Minor bug: this value was hard coded and it was not possible to overwrite it.

[mjsax]

The test fails here... (cf. over TODO)

[mjsax]

This is the workaround that make the test (clean run) pass. For the error injection run, the test passed w/ and w/o this partitioner.

[mjsax]

This is just needed to make the partitioner work for writing into input topics and to use within KS to write into output topic.

[guozhangwang]

Hmm, this sounds to me that the StreamProducer's own partitionsFor did not return the num.partitions so we ended up calling send with partition == null, since otherwise we will get the partition as

partition = partitioner.partition(topic, key, value, partitions.size());

where partitioner is the StreamsPartitioner and the producer's own partitioner should not be used.

[mjsax]

I don't think so. The original impl (just for the upstream producer to write into the input topics) was:

return ((Long) key).intValue() % NUM_TOPIC_PARTITIONS;

However, this assumes that key is of type Long what is not true when used within streams, because Streams does serialize all data upfront and key and value type is byte[] -- thus, we need to deserialize to get the original key object.

[guozhangwang]

What I was asking is for the necessity of

properties.put(StreamsConfig.producerPrefix(ProducerConfig.PARTITIONER_CLASS_CONFIG), KeyPartitioner.class);

As I mentioned, Streams has its own StreamsPartitioner, and if it can get the actual not-null partition value passing to the send call, then the embedded producer's partitioner would not be used. Maybe I missed something critical here --- did you mean this config is only used for sending data to the source topics? If yes why put it into a streams props?

[mjsax]

We use 6 clients now, to do some error injection in "mixed mode". To avoid JXM warnings, we cannot create all clients at the same time and thus cannot use try-with-resources...

[mjsax]

This is new: for the crash case, in inject more errors in mixed mode, after we called sendOffsetsToTransaction() but before actually committing.

[mjsax]

This is the second part of the "mixed mode" test: client1 is on eos-beta and client2 is on eoa-alpha. In the first part above, we crashed the second client and restarted it in eos-alpha mode. In this second part, we crash client1 and restart it in eos-beta mode.

The actual upgrade continues in the next phase.

[mjsax]

To get the correct observed state transition, we need to add them after waitForCondition failed... Originally, we just called:

waitForCondition(
                () -> observed.equals(expected),
                MAX_WAIT_TIME_MS,
                "Client did not startup on time." + observed
            );

but this creates the error string with observed when started and thus the error message is miss-leading and useless

[mjsax]

With 4 partitions, we need a custom partitioner to make sure we write the 4 different keys into 4 different partitions -- the default partitioner would only write data to 2 partitions (this behavior, ie, empty partitions vs. non-empty partitions, seems to be related to the bug when the test fails -- ensuring that data is written into all partitions avoids the issue).

[mjsax]

Updated this PR. Call for review.

[guozhangwang]

nit: Could you add the original comment explaining why we set it to smaller value too?

[guozhangwang]

Hmm, this sounds to me that the StreamProducer's own partitionsFor did not return the num.partitions so we ended up calling send with partition == null, since otherwise we will get the partition as

partition = partitioner.partition(topic, key, value, partitions.size());

where partitioner is the StreamsPartitioner and the producer's own partitioner should not be used.

[mjsax]

Java 8 passed.
Java 11: org.apache.kafka.streams.integration.GlobalKTableIntegrationTest.shouldKStreamGlobalKTableLeftJoin
Java 14:

kafka.api.PlaintextConsumerTest.testLowMaxFetchSizeForRequestAndPartition
org.apache.kafka.streams.integration.GlobalKTableIntegrationTest.shouldKStreamGlobalKTableJoin

Retest this please.

[mjsax]

Rebase this PR -- failed before, because the ProcessorContext "generics" commit was reverted.

[mjsax]

Java 11 passed.
Java 8:

org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[false]
org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest.shouldKStreamGlobalKTableJoin[exactly_once_beta]

Java 14: GlobalKTableEOSIntegrationTest.shouldKStreamGlobalKTableLeftJoin[exactly_once_beta]

[mjsax]

Retest this please

[mjsax]

@guozhangwang The failing test run has truncate logs. So it's hard to know what the root cause it. The high level "pattern" seems to be similar to what we have observed before: we are waiting for two rebalances (due to incremental rebalancing) but only get one... could not reproduce locally yet. \cc @ableegoldman

Btw: it seems that the test retry passed (https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2085/testReport/junit/org.apache.kafka.streams.integration/EosBetaUpgradeIntegrationTest/)

The build failed, because the was a second failing test.

[mjsax]

Java 14 passed.
Java 8: kafka.network.SocketServerTest.testClientDisconnectionUpdatesRequestMetrics
Java 11:

org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[true]
org.apache.kafka.streams.integration.EosIntegrationTest.shouldNotViolateEosIfOneTaskFailsWithState[exactly_once]

Merging this.

[abbccdda]

Great work! Finally see this last piece merged :)

[ableegoldman]

Java 11: org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[true]

@mjsax Should we use @ignore rather than merge known flaky tests while they're still under investigation? That's what we did for some of the KIP-441 tests (granted they were failing at almost 100%, but still 🙂)

[mjsax]

@ableegoldman Fair comment.

I discussed with @guozhangwang and we thought it would be ok to merge for now... If the test becomes a burden, we can also disable it. Did we see many PR build failures for it so far?