### Question:
KafkaSinkConnector extends SinkConnector { @Override public void start(Map<String, String> taskConfig) { config = new HashMap<>(); taskConfig.forEach((key, value) -> { if (key.startsWith(PRODUCER_PREFIX)) { config.put(key.substring(PRODUCER_PREFIX.length()), value); } }); config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, taskConfig.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)); config = Collections.unmodifiableMap(config); LOGGER.debug("Using {} for config", config); }  @Override String version(); @Override void start(Map<String, String> taskConfig); @Override Class<? extends Task> taskClass(); @Override List<Map<String, String>> taskConfigs(int totalConfigs); @Override void stop(); @Override ConfigDef config(); static final String PRODUCER_PREFIX; }### Answer:
@Test public void start() { connector.start(startConfig); assertThat(connector.config, is(taskConfig)); }