MINOR: Add synchronization to the protocol name check

[abbccdda]

Addressing the comment from #8324 (comment)

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[abbccdda]

@dajac Regarding the comment: #8324 (comment)
Could you elaborate more on what an explicit handling means? I only saw that we raised an error but was not handling it directly either.

[guozhangwang]

test this please

[guozhangwang]

@dajac just to clarify, are you concerning that the generation() may change between the check and the error-log? If yes maybe we do not need to synchronize the whole function, instead we just get a reference of the returned generation() call and use that in the error-log, since the generation object is immutable.

[dajac]

@guozhangwang Yes, keeping the reference is fine. I was concerned by the check itself because we were calling generation() twice in the previous implementation thus we could get two different instances.

generation() != Generation.NO_GENERATION && !protocolName.equals(generation().protocolName)

I haven't thought about the error-log but it is also a good point.

[guozhangwang]

test this please

[guozhangwang]

test this please

[dajac]

@dajac Regarding the comment: #8324 (comment)
Could you elaborate more on what an explicit handling means? I only saw that we raised an error but was not handling it directly either.

@abbccdda I was thinking about putting an explicit check of the generation is the response handler itself instead of having it hidden in the isProtocolNameInconsistent method. For instance, we could add something like this here.

Generation generationSnapshot = generation();
if (generation == Generation.NO_GENERATION) {
  future.raise(Errors.ILLEGAL_GENERATION);
} else if (isProtocolTypeInconsistent(syncResponse.data.protocolType())) {
  future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
} else if (isProtocolNameInconsistent(generationSnapshot, syncResponse.data.protocolName())) {
  future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
} else {
  sensors.syncSensor.record(response.requestLatencyMs());
  future.complete(ByteBuffer.wrap(syncResponse.data.assignment()));
}

This would make the fact the the generation can be reseted in the middle of the rebalance more explicit. Would this make sense?