[FLINK-35215][core] Fix the bug when Kryo serialize length is 0

[1996fanrui]

What is the purpose of the change

The performance of serializerKryo and serializerKryoWithoutRegistration are regressed, I checked recent commits, and found FLINK-34954 changed related logic.

Brief change log

• The first commit: Revert "[FLINK-34954][core] Kryo Input bug fix"
• The second commit: [FLINK-35215][core] Fix the bug when Kryo serialize length is 0
  • This PR reverts the FLINK-34954, and try to re-implement to avoid the performance regression.

[flinkbot]

CI report:

• ba2740f Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[reswqa]

Not to block this, just curious do you know why FLINK-34954 affects performance?

[1996fanrui]

Not to block this, just curious do you know why FLINK-34954 affects performance?

Hi @reswqa , I'm not very sure the reason. But I guess:

• For non-empty data, let us assume inputStream.read will be called once.
  • Before FLINK-34954, we only check bytesRead == count once.
  • After FLINK-34954, we call bytesRead < count twice.
• Also, I'm not sure whether FLINK-34954 breaks any JIT optimization.

[reswqa]

Thanks! Yes, I don't know the details also, but it's possible that JIT optimizations were affected.

[1996fanrui]

This if (count == 0) { is alternative change for FLINK-34954, I ran the demo that provided in FLINK-34954, it works well. So I think the change is fine.

Also, I ran the benchmark, the performance of serializerKryo and serializerKryoWithoutRegistration are recovered.

[1996fanrui]

Hi @dannycranmer , this issue is the blocker of Flink 1.20, and you are the reviewer of #24586, would you mind reviewing this PR as well? thanks in advance.

[uce]

@1996fanrui Can you add the demo from FLINK-34954 as a test?

[kkrugler]

I'm leaving in 2 days for a 4+ month backpack trip (hiking the PCT), otherwise what I'd do is try to hook up a test that uses JMH to accurately assess the impact of the change. To respond to the particular comment on why performance changed:

we call bytesRead < count twice

The overhead of two comparisons, versus even one call to inputStream.read, should be miniscule.

I'm not sure whether FLINK-34954 breaks any JIT optimization

If a JIT optimization would result in skipping calls to inputStream.read, then yes that could have an impact. But that would be a change in logic.

The one thing I see is in the change to readBytes(), where if it's often called with a count of 0, then your change skips the try/catch block, which is non-trivial. But I thought the fix was to avoid a failure for this exact case, so it shouldn't be happening in the current benchmark code.

[1996fanrui]

I'm leaving in 2 days for a 4+ month backpack trip (hiking the PCT)

Wow, have a nice trip!

To respond to the particular comment on why performance changed:

we call bytesRead < count twice

The overhead of two comparisons, versus even one call to inputStream.read, should be miniscule.

I'm not sure whether FLINK-34954 breaks any JIT optimization

If a JIT optimization would result in skipping calls to inputStream.read, then yes that could have an impact. But that would be a change in logic.

As I said before, I'm not sure the reason, and both of them are my guess.

The one thing I see is in the change to readBytes(), where if it's often called with a count of 0, then your change skips the try/catch block, which is non-trivial. But I thought the fix was to avoid a failure for this exact case, so it shouldn't be happening in the current benchmark code.

Sorry, I didn't fully understand this sentence. Do you mean skip the try/catch block is not fine?

[kkrugler]

Sorry for the confusing sentence. I meant that skipping the try/catch block would save a lot of cycles, but your change that causes it to be (correctly) skipped when the count is 0 shouldn't be the common case, so I wouldn't expect that to impact the benchmark results.

[1996fanrui]

Sorry for the confusing sentence. I meant that skipping the try/catch block would save a lot of cycles, but your change that causes it to be (correctly) skipped when the count is 0 shouldn't be the common case, so I wouldn't expect that to impact the benchmark results.

Thanks for the quick clarification! Yes, my change only adds a simple check if (count == 0) return;, and it doesn't impact performance after I run the benchmark.

[pnowojski]

I'm leaving in 2 days for a 4+ month backpack trip (hiking the PCT), otherwise what I'd do is try to hook up a test that uses JMH to accurately assess the impact of the change. To respond to the particular comment on why performance changed:

we call bytesRead < count twice

The overhead of two comparisons, versus even one call to inputStream.read, should be miniscule.

I agree that some explanation is missing here. While I would be also curious to understand all of this better, I'm fine if this is merged as is. In the absence of a better performance indicator, I would relay on our imperfect benchmarking results, which suggests we should merge this one over the current master.

We can also continue investigation asynchronously after merging this fix if someone is interested in doing that.

[1996fanrui]

Thanks @pnowojski for the comment and review!

Let me merge this PR first, and I will follow the benchmark in the following days.