@Test public void connectorStatus() { ConnectorTaskId taskId = new ConnectorTaskId(connector, 0); ConfigBackingStore configStore = strictMock(ConfigBackingStore.class); StatusBackingStore statusStore = strictMock(StatusBackingStore.class); AbstractHerder herder = partialMockBuilder(AbstractHerder.class) .withConstructor(Worker.class, String.class, StatusBackingStore.class, ConfigBackingStore.class) .withArgs(worker, workerId, statusStore, configStore) .addMockedMethod("generation") .createMock(); EasyMock.expect(herder.generation()).andStubReturn(generation); EasyMock.expect(statusStore.get(connector)) .andReturn(new ConnectorStatus(connector, AbstractStatus.State.RUNNING, workerId, generation)); EasyMock.expect(statusStore.getAll(connector)) .andReturn(Collections.singletonList( new TaskStatus(taskId, AbstractStatus.State.UNASSIGNED, workerId, generation))); replayAll(); ConnectorStateInfo state = herder.connectorStatus(connector); assertEquals(connector, state.name()); assertEquals("RUNNING", state.connector().state()); assertEquals(1, state.tasks().size()); assertEquals(workerId, state.connector().workerId()); ConnectorStateInfo.TaskState taskState = state.tasks().get(0); assertEquals(0, taskState.id()); assertEquals("UNASSIGNED", taskState.state()); assertEquals(workerId, taskState.workerId()); verifyAll(); }
@Override public ConnectorStateInfo connectorStatus(String connName) { ConnectorStatus connector = statusBackingStore.get(connName); if (connector == null) throw new NotFoundException("No status found for connector " + connName); Collection<TaskStatus> tasks = statusBackingStore.getAll(connName); ConnectorStateInfo.ConnectorState connectorState = new ConnectorStateInfo.ConnectorState( connector.state().toString(), connector.workerId(), connector.trace()); List<ConnectorStateInfo.TaskState> taskStates = new ArrayList<>(); for (TaskStatus status : tasks) { taskStates.add(new ConnectorStateInfo.TaskState(status.id().task(), status.state().toString(), status.workerId(), status.trace())); } Collections.sort(taskStates); return new ConnectorStateInfo(connName, connectorState, taskStates); }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConnectorStateInfo connectorStatus(String connName) { ConnectorStatus connector = statusBackingStore.get(connName); if (connector == null) throw new NotFoundException("No status found for connector " + connName); Collection<TaskStatus> tasks = statusBackingStore.getAll(connName); ConnectorStateInfo.ConnectorState connectorState = new ConnectorStateInfo.ConnectorState( connector.state().toString(), connector.workerId(), connector.trace()); List<ConnectorStateInfo.TaskState> taskStates = new ArrayList<>(); for (TaskStatus status : tasks) { taskStates.add(new ConnectorStateInfo.TaskState(status.id().task(), status.state().toString(), status.workerId(), status.trace())); } Collections.sort(taskStates); return new ConnectorStateInfo(connName, connectorState, taskStates); } }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConnectorStateInfo connectorStatus(String connName) { ConnectorStatus connector = statusBackingStore.get(connName); if (connector == null) throw new NotFoundException("No status found for connector " + connName); Collection<TaskStatus> tasks = statusBackingStore.getAll(connName); ConnectorStateInfo.ConnectorState connectorState = new ConnectorStateInfo.ConnectorState( connector.state().toString(), connector.workerId(), connector.trace()); List<ConnectorStateInfo.TaskState> taskStates = new ArrayList<>(); for (TaskStatus status : tasks) { taskStates.add(new ConnectorStateInfo.TaskState(status.id().task(), status.state().toString(), status.workerId(), status.trace())); } Collections.sort(taskStates); return new ConnectorStateInfo(connName, connectorState, taskStates); } AbstractHerder(Worker worker,
                          String workerId,
                          StatusBackingStore statusBackingStore,
                          ConfigBackingStore configBackingStore); }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConnectorStateInfo connectorStatus(String connName) { ConnectorStatus connector = statusBackingStore.get(connName); if (connector == null) throw new NotFoundException("No status found for connector " + connName); Collection<TaskStatus> tasks = statusBackingStore.getAll(connName); ConnectorStateInfo.ConnectorState connectorState = new ConnectorStateInfo.ConnectorState( connector.state().toString(), connector.workerId(), connector.trace()); List<ConnectorStateInfo.TaskState> taskStates = new ArrayList<>(); for (TaskStatus status : tasks) { taskStates.add(new ConnectorStateInfo.TaskState(status.id().task(), status.state().toString(), status.workerId(), status.trace())); } Collections.sort(taskStates); return new ConnectorStateInfo(connName, connectorState, taskStates); } AbstractHerder(Worker worker,
                          String workerId,
                          StatusBackingStore statusBackingStore,
                          ConfigBackingStore configBackingStore); @Override void onStartup(String connector); @Override void onPause(String connector); @Override void onResume(String connector); @Override void onShutdown(String connector); @Override void onFailure(String connector, Throwable cause); @Override void onStartup(ConnectorTaskId id); @Override void onFailure(ConnectorTaskId id, Throwable cause); @Override void onShutdown(ConnectorTaskId id); @Override void onResume(ConnectorTaskId id); @Override void onPause(ConnectorTaskId id); @Override void onDeletion(String connector); @Override void pauseConnector(String connector); @Override void resumeConnector(String connector); @Override Plugins plugins(); @Override ConnectorStateInfo connectorStatus(String connName); @Override ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id); @Override ConfigInfos validateConnectorConfig(Map<String, String> connectorProps); static ConfigInfos generateResult(String connType, Map<String, ConfigKey> configKeys, List<ConfigValue> configValues, List<String> groups); }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConnectorStateInfo connectorStatus(String connName) { ConnectorStatus connector = statusBackingStore.get(connName); if (connector == null) throw new NotFoundException("No status found for connector " + connName); Collection<TaskStatus> tasks = statusBackingStore.getAll(connName); ConnectorStateInfo.ConnectorState connectorState = new ConnectorStateInfo.ConnectorState( connector.state().toString(), connector.workerId(), connector.trace()); List<ConnectorStateInfo.TaskState> taskStates = new ArrayList<>(); for (TaskStatus status : tasks) { taskStates.add(new ConnectorStateInfo.TaskState(status.id().task(), status.state().toString(), status.workerId(), status.trace())); } Collections.sort(taskStates); return new ConnectorStateInfo(connName, connectorState, taskStates); } AbstractHerder(Worker worker,
                          String workerId,
                          StatusBackingStore statusBackingStore,
                          ConfigBackingStore configBackingStore); @Override void onStartup(String connector); @Override void onPause(String connector); @Override void onResume(String connector); @Override void onShutdown(String connector); @Override void onFailure(String connector, Throwable cause); @Override void onStartup(ConnectorTaskId id); @Override void onFailure(ConnectorTaskId id, Throwable cause); @Override void onShutdown(ConnectorTaskId id); @Override void onResume(ConnectorTaskId id); @Override void onPause(ConnectorTaskId id); @Override void onDeletion(String connector); @Override void pauseConnector(String connector); @Override void resumeConnector(String connector); @Override Plugins plugins(); @Override ConnectorStateInfo connectorStatus(String connName); @Override ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id); @Override ConfigInfos validateConnectorConfig(Map<String, String> connectorProps); static ConfigInfos generateResult(String connType, Map<String, ConfigKey> configKeys, List<ConfigValue> configValues, List<String> groups);  }