docs: @SuppressWarnings("unchecked")
    public void emitLogsAndMetrics(
        @Nullable final Throwable e,
        @Nullable final String remoteAddress,
        final long bytesWritten
    )
    {
      if (baseQuery == null) {
        // Never initialized, don't log or emit anything.
        return;
      }
  
      if (state == State.DONE) {
        log.warn("Tried to emit logs and metrics twice for query[%s]!", baseQuery.getId());
      }
  
      state = State.DONE;
  
      final boolean success = e == null;
  
      try {
        final long queryTimeNs = System.nanoTime() - startNs;
  
        QueryMetrics queryMetrics = DruidMetrics.makeRequestMetrics(
            queryMetricsFactory,
            toolChest,
            baseQuery,
            StringUtils.nullToEmptyNonDruidDataString(remoteAddress)
        );
        queryMetrics.success(success);
        queryMetrics.reportQueryTime(queryTimeNs);
  
        if (bytesWritten >= 0) {
          queryMetrics.reportQueryBytes(bytesWritten);
        }
  
        if (authenticationResult != null) {
          queryMetrics.identity(authenticationResult.getIdentity());
        }
  
        queryMetrics.emit(emitter);
  
        final Map<String, Object> statsMap = new LinkedHashMap<>();
        statsMap.put("query/time", TimeUnit.NANOSECONDS.toMillis(queryTimeNs));
        statsMap.put("query/bytes", bytesWritten);
        statsMap.put("success", success);
  
        if (authenticationResult != null) {
          statsMap.put("identity", authenticationResult.getIdentity());
        }
  
        if (e != null) {
          statsMap.put("exception", e.toString());
  
          if (e instanceof QueryInterruptedException) {
            // Mimic behavior from QueryResource, where this code was originally taken from.
            log.warn(e, "Exception while processing queryId [%s]", baseQuery.getId());
            statsMap.put("interrupted", true);
            statsMap.put("reason", e.toString());
          }
        }
        requestLogger.logNativeQuery(
            RequestLogLine.forNative(
                baseQuery,
                DateTimes.utc(startMs),
                StringUtils.nullToEmptyNonDruidDataString(remoteAddress),
                new QueryStats(statsMap)
            )
        );
      }
      catch (Exception ex) {
        log.error(ex, "Unable to log query [%s]!", baseQuery);
      }
    }