KafkaBasedConnectorConfig { public Properties getConsumerProps() { Properties consumerProps = new Properties(); consumerProps.putAll(_consumerProps); return consumerProps; } KafkaBasedConnectorConfig(Properties properties); String getDefaultKeySerde(); String getDefaultValueSerde(); long getCommitIntervalMillis(); Duration getCommitTimeout(); long getPollTimeoutMillis(); int getRetryCount(); Duration getRetrySleepDuration(); boolean getPausePartitionOnError(); Duration getPauseErrorPartitionDuration(); boolean getEnablePollDurationMillisMetric(); Properties getConsumerProps(); KafkaConsumerFactory<?, ?> getConsumerFactory(); VerifiableProperties getConnectorProps(); int getDaemonThreadIntervalSeconds(); long getNonGoodStateThresholdMillis(); long getProcessingDelayLogThresholdMillis(); boolean getEnablePartitionAssignment(); static final String DOMAIN_KAFKA_CONSUMER; static final String CONFIG_COMMIT_INTERVAL_MILLIS; static final String CONFIG_COMMIT_TIMEOUT_MILLIS; static final String CONFIG_POLL_TIMEOUT_MILLIS; static final String CONFIG_CONSUMER_FACTORY_CLASS; static final String CONFIG_DEFAULT_KEY_SERDE; static final String CONFIG_DEFAULT_VALUE_SERDE; static final String CONFIG_RETRY_COUNT; static final String CONFIG_RETRY_SLEEP_DURATION_MILLIS; static final String CONFIG_PAUSE_PARTITION_ON_ERROR; static final String CONFIG_PAUSE_ERROR_PARTITION_DURATION_MILLIS; static final String CONFIG_ENABLE_POLL_DURATION_MILLIS_METRIC; static final String DAEMON_THREAD_INTERVAL_SECONDS; static final String NON_GOOD_STATE_THRESHOLD_MILLIS; static final String PROCESSING_DELAY_LOG_THRESHOLD_MILLIS; static final String ENABLE_PARTITION_ASSIGNMENT; static final long DEFAULT_NON_GOOD_STATE_THRESHOLD_MILLIS; static final long MIN_NON_GOOD_STATE_THRESHOLD_MILLIS; }
@Test public void testConsumerConfig() { Properties props = new Properties(); props.put("consumer.someProperty", "someValue"); props.put("consumer.someProperty1", "someValue1"); props.put("consumer.someProperty2", "someValue2"); KafkaBasedConnectorConfig config = new KafkaBasedConnectorConfig(props); Properties configProps = config.getConsumerProps(); Properties copiedProps = new Properties(); copiedProps.putAll(configProps); Assert.assertNotNull(copiedProps); Assert.assertEquals(copiedProps.getProperty("someProperty"), "someValue"); Assert.assertEquals(copiedProps.getProperty("someProperty1"), "someValue1"); Assert.assertEquals(copiedProps.getProperty("someProperty2"), "someValue2"); Properties taskProps = KafkaConnectorTask.getKafkaConsumerProperties(configProps, "groupId", KafkaConnectionString.valueOf("kafkassl: Assert.assertEquals(taskProps.getProperty("someProperty"), "someValue"); Assert.assertEquals(taskProps.getProperty("someProperty1"), "someValue1"); Assert.assertEquals(taskProps.getProperty("someProperty2"), "someValue2"); }