@Test(expected = ConnectException.class) public void testMissingTopic() throws InterruptedException { replay(); config.remove(FileStreamSourceConnector.TOPIC_CONFIG); task.start(config); }
@Override public void start(Map<String, String> props) { filename = props.get(FileStreamSourceConnector.FILE_CONFIG); if (filename == null || filename.isEmpty()) { stream = System.in; streamOffset = null; reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8)); } topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG); if (topic == null) throw new ConnectException("FileStreamSourceTask config missing topic setting"); }
FileStreamSourceTask extends SourceTask { @Override public void start(Map<String, String> props) { filename = props.get(FileStreamSourceConnector.FILE_CONFIG); if (filename == null || filename.isEmpty()) { stream = System.in; streamOffset = null; reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8)); } topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG); if (topic == null) throw new ConnectException("FileStreamSourceTask config missing topic setting"); } }
FileStreamSourceTask extends SourceTask { @Override public void start(Map<String, String> props) { filename = props.get(FileStreamSourceConnector.FILE_CONFIG); if (filename == null || filename.isEmpty()) { stream = System.in; streamOffset = null; reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8)); } topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG); if (topic == null) throw new ConnectException("FileStreamSourceTask config missing topic setting"); }  }
FileStreamSourceTask extends SourceTask { @Override public void start(Map<String, String> props) { filename = props.get(FileStreamSourceConnector.FILE_CONFIG); if (filename == null || filename.isEmpty()) { stream = System.in; streamOffset = null; reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8)); } topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG); if (topic == null) throw new ConnectException("FileStreamSourceTask config missing topic setting"); }  @Override String version(); @Override void start(Map<String, String> props); @Override List<SourceRecord> poll(); @Override void stop(); }
FileStreamSourceTask extends SourceTask { @Override public void start(Map<String, String> props) { filename = props.get(FileStreamSourceConnector.FILE_CONFIG); if (filename == null || filename.isEmpty()) { stream = System.in; streamOffset = null; reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8)); } topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG); if (topic == null) throw new ConnectException("FileStreamSourceTask config missing topic setting"); }  @Override String version(); @Override void start(Map<String, String> props); @Override List<SourceRecord> poll(); @Override void stop(); static final String FILENAME_FIELD; static final String POSITION_FIELD; }