Repository: apache/kafka
File: core/src/main/scala/kafka/server/KafkaApis.scala (Scala)

Code before:
        } else {
          val responseData = response.data()
          val topicsToCreate = response.creatableTopics().asScala
          if (topicsToCreate.nonEmpty) {

            val createTopicUnauthorized =
              if(!authHelper.authorize(request.context, CREATE, CLUSTER, CLUSTER_NAME, logIfDenied = false))
                authHelper.partitionSeqByAuthorized(request.context, CREATE, TOPIC, topicsToCreate.keys.toSeq)(identity[String])._2
              else Set.empty

            if (createTopicUnauthorized.nonEmpty) {
              if (responseData.status() == null) {
                responseData.setStatus(new util.ArrayList());
              }
              val missingInternalTopicStatus =
                responseData.status().stream().filter(x => x.statusCode() == StreamsGroupHeartbeatResponse.Status.MISSING_INTERNAL_TOPICS.code()).findFirst()
              if (missingInternalTopicStatus.isPresent) {
                missingInternalTopicStatus.get().setStatusDetail(
                  missingInternalTopicStatus.get().statusDetail() + "; Unauthorized to CREATE on topics " + createTopicUnauthorized.mkString(",") + "."
                )
              } else {
                responseData.status().add(
                  new StreamsGroupHeartbeatResponseData.Status()
                    .setStatusCode(StreamsGroupHeartbeatResponse.Status.MISSING_INTERNAL_TOPICS.code())
                    .setStatusDetail("Unauthorized to CREATE on topics " + createTopicUnauthorized.mkString(",") + ".")
                )
              }
            } else {
              autoTopicCreationManager.createStreamsInternalTopics(topicsToCreate, requestContext);
            }
          }

          requestHelper.sendMaybeThrottle(request, new StreamsGroupHeartbeatResponse(responseData))
        }
      }
    }
  }

  def handleStreamsGroupDescribe(request: RequestChannel.Request): CompletableFuture[Unit] = {
    val streamsGroupDescribeRequest = request.body[StreamsGroupDescribeRequest]
    val includeAuthorizedOperations = streamsGroupDescribeRequest.data.includeAuthorizedOperations

    if (!isStreamsGroupProtocolEnabled) {
      // The API is not enabled by default. If it is not enabled, we fail directly here.
      requestHelper.sendMaybeThrottle(request, request.body[StreamsGroupDescribeRequest].getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
      CompletableFuture.completedFuture[Unit](())
    } else {
      val response = new StreamsGroupDescribeResponseData()

      val authorizedGroups = new ArrayBuffer[String]()
      streamsGroupDescribeRequest.data.groupIds.forEach { groupId =>

Review comment (security):
The code always appends an 'Unauthorized to CREATE' detail when a missing-internal-topics status exists, even if `createTopicUnauthorized` is empty. This will leave a dangling message when all CREATE ACLs are present and break the corresponding test. Wrap the append logic in a check for nonEmpty `createTopicUnauthorized`.
```suggestion
              if (createTopicUnauthorized.nonEmpty) {
                if (missingInternalTopicStatus.isPresent) {
                  missingInternalTopicStatus.get().setStatusDetail(
                    missingInternalTopicStatus.get().statusDetail() + "; Unauthorized to CREATE on topics " + createTopicUnauthorized.mkString(",") + "."
                  )
                } else {
                  responseData.status().add(
                    new StreamsGroupHeartbeatResponseData.Status()
                      .setStatusCode(StreamsGroupHeartbeatResponse.Status.MISSING_INTERNAL_TOPICS.code())
                      .setStatusDetail("Unauthorized to CREATE on topics " + createTopicUnauthorized.mkString(",") + ".")
                  )
                }
```

Code after:
        } else {
          val responseData = response.data()
          val topicsToCreate = response.creatableTopics().asScala
          if (topicsToCreate.nonEmpty) {

            val createTopicUnauthorized =
              if(!authHelper.authorize(request.context, CREATE, CLUSTER, CLUSTER_NAME, logIfDenied = false))
                authHelper.partitionSeqByAuthorized(request.context, CREATE, TOPIC, topicsToCreate.keys.toSeq)(identity[String])._2
              else Set.empty

            if (createTopicUnauthorized.nonEmpty) {
              if (responseData.status() == null) {
                responseData.setStatus(new util.ArrayList());
              }
              val missingInternalTopicStatus =
                responseData.status().stream().filter(x => x.statusCode() == StreamsGroupHeartbeatResponse.Status.MISSING_INTERNAL_TOPICS.code()).findFirst()
              if (missingInternalTopicStatus.isPresent) {
                missingInternalTopicStatus.get().setStatusDetail(
                  missingInternalTopicStatus.get().statusDetail() + "; Unauthorized to CREATE on topics " + createTopicUnauthorized.mkString(", ") + "."
                )
              } else {
                responseData.status().add(
                  new StreamsGroupHeartbeatResponseData.Status()
                    .setStatusCode(StreamsGroupHeartbeatResponse.Status.MISSING_INTERNAL_TOPICS.code())
                    .setStatusDetail("Unauthorized to CREATE on topics " + createTopicUnauthorized.mkString(", ") + ".")
                )
              }
            } else {
              autoTopicCreationManager.createStreamsInternalTopics(topicsToCreate, requestContext);
            }
          }

          requestHelper.sendMaybeThrottle(request, new StreamsGroupHeartbeatResponse(responseData))
        }
      }
    }
  }

  def handleStreamsGroupDescribe(request: RequestChannel.Request): CompletableFuture[Unit] = {
    val streamsGroupDescribeRequest = request.body[StreamsGroupDescribeRequest]
    val includeAuthorizedOperations = streamsGroupDescribeRequest.data.includeAuthorizedOperations

    if (!isStreamsGroupProtocolEnabled) {
      // The API is not enabled by default. If it is not enabled, we fail directly here.
      requestHelper.sendMaybeThrottle(request, request.body[StreamsGroupDescribeRequest].getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
      CompletableFuture.completedFuture[Unit](())
    } else {
      val response = new StreamsGroupDescribeResponseData()

      val authorizedGroups = new ArrayBuffer[String]()
      streamsGroupDescribeRequest.data.groupIds.forEach { groupId =>