code: @Override
      public byte[] getKey() {
          StringBuilder sb = new StringBuilder(128);
          sb.append(clusterName).append(MetricUtils.AT).append(topologyId).append(MetricUtils.AT)
                  .append(taskId).append(MetricUtils.AT);
          if (start != null) {
              sb.append(start.getTime());
          } else {
              sb.append(end.getTime());
          }
          return sb.toString().getBytes();
      }
docs: key: clusterName + topologyId + taskId + time