public InternalOperatorMetricGroup getOrAddOperator(String operatorName) {
        return getOrAddOperator(OperatorID.fromJobVertexID(vertexId), operatorName);
    }
@Test
    void testOperatorNameTruncation() throws Exception {
        Configuration cfg = new Configuration();
        cfg.set(MetricOptions.SCOPE_NAMING_OPERATOR, ScopeFormat.SCOPE_OPERATOR_NAME);
        MetricRegistryImpl registry =
                new MetricRegistryImpl(MetricRegistryTestUtils.fromConfiguration(cfg));
        TaskManagerMetricGroup tm =
                TaskManagerMetricGroup.createTaskManagerMetricGroup(
                        registry, "host", new ResourceID("id"));

        TaskMetricGroup taskMetricGroup =
                tm.addJob(new JobID(), "jobname").addTask(createExecutionAttemptId(), "task");

        String originalName = new String(new char[100]).replace("\0", "-");
        InternalOperatorMetricGroup operatorMetricGroup =
                taskMetricGroup.getOrAddOperator(originalName);

        String storedName = operatorMetricGroup.getScopeComponents()[0];
        assertThat(storedName.length()).isEqualTo(TaskMetricGroup.METRICS_OPERATOR_NAME_MAX_LENGTH);
        assertThat(originalName.substring(0, TaskMetricGroup.METRICS_OPERATOR_NAME_MAX_LENGTH))
                .isEqualTo(storedName);
        registry.closeAsync().get();
    }