@Test public void taskStatus() { ConnectorTaskId taskId = new ConnectorTaskId("connector", 0); String workerId = "workerId"; 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(5); final Capture<TaskStatus> statusCapture = EasyMock.newCapture(); statusStore.putSafe(EasyMock.capture(statusCapture)); EasyMock.expectLastCall(); EasyMock.expect(statusStore.get(taskId)).andAnswer(new IAnswer<TaskStatus>() { @Override public TaskStatus answer() throws Throwable { return statusCapture.getValue(); } }); replayAll(); herder.onFailure(taskId, new RuntimeException()); ConnectorStateInfo.TaskState taskState = herder.taskStatus(taskId); assertEquals(workerId, taskState.workerId()); assertEquals("FAILED", taskState.state()); assertEquals(0, taskState.id()); assertNotNull(taskState.trace()); verifyAll(); }
@Override public ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id) { TaskStatus status = statusBackingStore.get(id); if (status == null) throw new NotFoundException("No status found for task " + id); return new ConnectorStateInfo.TaskState(id.task(), status.state().toString(), status.workerId(), status.trace()); }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id) { TaskStatus status = statusBackingStore.get(id); if (status == null) throw new NotFoundException("No status found for task " + id); return new ConnectorStateInfo.TaskState(id.task(), status.state().toString(), status.workerId(), status.trace()); } }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id) { TaskStatus status = statusBackingStore.get(id); if (status == null) throw new NotFoundException("No status found for task " + id); return new ConnectorStateInfo.TaskState(id.task(), status.state().toString(), status.workerId(), status.trace()); } AbstractHerder(Worker worker,
                          String workerId,
                          StatusBackingStore statusBackingStore,
                          ConfigBackingStore configBackingStore); }
AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener { @Override public ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id) { TaskStatus status = statusBackingStore.get(id); if (status == null) throw new NotFoundException("No status found for task " + id); return new ConnectorStateInfo.TaskState(id.task(), status.state().toString(), status.workerId(), status.trace()); } 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.TaskState taskStatus(ConnectorTaskId id) { TaskStatus status = statusBackingStore.get(id); if (status == null) throw new NotFoundException("No status found for task " + id); return new ConnectorStateInfo.TaskState(id.task(), status.state().toString(), status.workerId(), status.trace()); } 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);  }