dajac commented on code in PR #14589:
URL: https://github.com/apache/kafka/pull/14589#discussion_r1371818152
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -310,16 +313,22 @@ public CompletableFuture<JoinGroupResponseData> joinGroup(
JoinGroupRequestData request,
BufferSupplier bufferSupplier
) {
+ CompletableFuture<JoinGroupResponseData> responseFuture = new
CompletableFuture<>();
+
if (!isActive.get()) {
- return
FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
- }
+ responseFuture.complete(new JoinGroupResponseData()
+ .setMemberId(request.memberId())
+ .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
+ );
- CompletableFuture<JoinGroupResponseData> responseFuture = new
CompletableFuture<>();
+ return responseFuture;
+ }
if (!isGroupIdNotEmpty(request.groupId())) {
responseFuture.complete(new JoinGroupResponseData()
Review Comment:
We could also use `return CompletableFuture.completedFuture` here.
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4341,6 +4371,15 @@ class KafkaApisTest {
false
)).thenReturn(group3Future)
+ val group4Future = new
CompletableFuture[OffsetFetchResponseData.OffsetFetchResponseGroup]()
Review Comment:
I also feel like that we don't fully repro the issue in this one. We also
need to test for both the all topics (null) and specific topics (non null)
cases. Do you agree?
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -3267,6 +3267,35 @@ class KafkaApisTest {
assertEquals(Errors.GROUP_ID_NOT_FOUND,
Errors.forCode(response.data.errorCode))
}
+ @Test
+ def testOffsetDeleteWithInvalidGroupWithTopLevelError(): Unit = {
Review Comment:
I think that we don't fully reproduce the bug with this test because `if
(data.topics().isEmpty()) {` would be true in this case as well. Would it be
possible to have a test which really have partial results awaiting in the
builder?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -310,16 +313,22 @@ public CompletableFuture<JoinGroupResponseData> joinGroup(
JoinGroupRequestData request,
BufferSupplier bufferSupplier
) {
+ CompletableFuture<JoinGroupResponseData> responseFuture = new
CompletableFuture<>();
+
if (!isActive.get()) {
- return
FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
- }
+ responseFuture.complete(new JoinGroupResponseData()
Review Comment:
nit: Could we also use `return CompletableFuture.completedFuture`)?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -139,7 +138,7 @@ public void testStartupShutdown() throws Exception {
}
@Test
- public void testConsumerGroupHeartbeatWhenNotStarted() {
+ public void testConsumerGroupHeartbeatWhenNotStarted() throws
ExecutionException, InterruptedException {
Review Comment:
Is it worth adding a similar test for each API? That would ensure that we
keep the contract.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -268,7 +269,9 @@ public
CompletableFuture<ConsumerGroupHeartbeatResponseData> consumerGroupHeartb
ConsumerGroupHeartbeatRequestData request
Review Comment:
Should we also update the javadoc of all those methods to specify the
contract a little more? I known that the old coordinator won't fully comply
with this but as it will go away, it should not be a big deal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
