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

Code before:
          requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(exception))
        } 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]()

Review comment (security):
[nitpick] Use `mkString(", ")` instead of `mkString(",")` to include a space after commas for readability in the statusDetail.
```suggestion
                  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:
          requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(exception))
        } 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]()