@Test public void testJobParamOverrides() { Context context = mock(Context.class); String testTopic = "topic123"; String testConsumerGroupId = "testKafkaConsumer-1"; Parameters params = ParameterTestUtils.createParameters(KafkaSourceParameters.TOPIC, testTopic, KafkaSourceParameters.PREFIX + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest", KafkaSourceParameters.PREFIX + ConsumerConfig.GROUP_ID_CONFIG, testConsumerGroupId, KafkaSourceParameters.PREFIX + ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 500); when(context.getParameters()).then((Answer<Parameters>) invocation -> params); MantisKafkaConsumerConfig mantisKafkaConsumerConfig = new MantisKafkaConsumerConfig(context); Map<String, Object> consumerProperties = mantisKafkaConsumerConfig.getConsumerProperties(); assertFalse(ConsumerConfig.configNames().contains(KafkaSourceParameters.TOPIC)); assertFalse(consumerProperties.containsKey(KafkaSourceParameters.TOPIC)); assertEquals("earliest", consumerProperties.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)); assertEquals(testConsumerGroupId, consumerProperties.get(ConsumerConfig.GROUP_ID_CONFIG)); assertEquals(500, consumerProperties.get(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG)); }
public Map<String, Object> getConsumerProperties() { return values().entrySet().stream() .filter(x -> x.getKey() != null && x.getValue() != null) .collect(Collectors.toMap(x -> x.getKey(), x -> (Object) x.getValue())); }
MantisKafkaConsumerConfig extends ConsumerConfig { public Map<String, Object> getConsumerProperties() { return values().entrySet().stream() .filter(x -> x.getKey() != null && x.getValue() != null) .collect(Collectors.toMap(x -> x.getKey(), x -> (Object) x.getValue())); } }
MantisKafkaConsumerConfig extends ConsumerConfig { public Map<String, Object> getConsumerProperties() { return values().entrySet().stream() .filter(x -> x.getKey() != null && x.getValue() != null) .collect(Collectors.toMap(x -> x.getKey(), x -> (Object) x.getValue())); } MantisKafkaConsumerConfig(Map<String, Object> props,
                                     Context context);  MantisKafkaConsumerConfig(Context context); }
MantisKafkaConsumerConfig extends ConsumerConfig { public Map<String, Object> getConsumerProperties() { return values().entrySet().stream() .filter(x -> x.getKey() != null && x.getValue() != null) .collect(Collectors.toMap(x -> x.getKey(), x -> (Object) x.getValue())); } MantisKafkaConsumerConfig(Map<String, Object> props,
                                     Context context);  MantisKafkaConsumerConfig(Context context); static Map<String, Object> defaultProps(); static List<ParameterDefinition<?>> getJobParameterDefinitions(); String getConsumerConfigStr(); Map<String, Object> getConsumerProperties(); }
MantisKafkaConsumerConfig extends ConsumerConfig { public Map<String, Object> getConsumerProperties() { return values().entrySet().stream() .filter(x -> x.getKey() != null && x.getValue() != null) .collect(Collectors.toMap(x -> x.getKey(), x -> (Object) x.getValue())); } MantisKafkaConsumerConfig(Map<String, Object> props,
                                     Context context);  MantisKafkaConsumerConfig(Context context); static Map<String, Object> defaultProps(); static List<ParameterDefinition<?>> getJobParameterDefinitions(); String getConsumerConfigStr(); Map<String, Object> getConsumerProperties(); static final String DEFAULT_AUTO_OFFSET_RESET; static final String DEFAULT_AUTO_COMMIT_ENABLED; static final String DEFAULT_BOOTSTRAP_SERVERS_CONFIG; static final int DEFAULT_AUTO_COMMIT_INTERVAL_MS; static final int DEFAULT_HEARTBEAT_INTERVAL_MS; static final int DEFAULT_SESSION_TIMEOUT_MS; static final int DEFAULT_FETCH_MIN_BYTES; static final int DEFAULT_FETCH_MAX_WAIT_MS; static final int DEFAULT_REQUEST_TIMEOUT_MS; static final int DEFAULT_CHECKPOINT_INTERVAL_MS; static final int DEFAULT_MAX_POLL_INTERVAL_MS; static final int DEFAULT_MAX_POLL_RECORDS; static final int DEFAULT_MAX_PARTITION_FETCH_BYTES; static final int DEFAULT_RECEIVE_BUFFER_BYTES; static final int DEFAULT_SEND_BUFFER_BYTES; static final Class<StringDeserializer> DEFAULT_KEY_DESERIALIZER; static final Class<ByteArrayDeserializer> DEFAULT_VALUE_DESERIALIZER; }