public long getAttempt() throws ResourceManagerException {
        Matcher matcher = TASK_MANAGER_POD_PATTERN.matcher(resourceID.toString());
        if (matcher.find()) {
            return Long.parseLong(matcher.group(1));
        } else {
            throw new ResourceManagerException(
                    "Error to parse KubernetesWorkerNode from " + resourceID + ".");
        }
    }
@Test
    public void testGetAttemptFromWorkerNode() throws Exception {
        final String correctPodName = "flink-on-kubernetes-taskmanager-11-5";
        final KubernetesWorkerNode workerNode =
                new KubernetesWorkerNode(new ResourceID(correctPodName));
        assertThat(workerNode.getAttempt()).isEqualTo(11L);

        final String wrongPodName = "flink-on-kubernetes-xxxtaskmanager-11-5";
        final KubernetesWorkerNode workerNode1 =
                new KubernetesWorkerNode(new ResourceID(wrongPodName));
        assertThatThrownBy(() -> workerNode1.getAttempt())
                .isInstanceOf(Exception.class)
                .hasMessageContaining(
                        "Error to parse KubernetesWorkerNode from " + wrongPodName + ".");
    }