### Question:
MantisKafkaProducerConfig extends ProducerConfig { public Map<String, Object> getProducerProperties() { return values().entrySet().stream() .filter(x -> x.getKey() != null && x.getValue() != null) .collect(Collectors.toMap(x -> x.getKey(), x -> (Object) x.getValue())); } MantisKafkaProducerConfig(Map<String, Object> props,
                                     Context context);  MantisKafkaProducerConfig(Context context); static Map<String, Object> defaultProps(); Map<String, Object> getProducerProperties(); static List<ParameterDefinition<?>> getJobParameterDefinitions(); static final String DEFAULT_BOOTSTRAP_SERVERS_CONFIG; static final String DEFAULT_ACKS_CONFIG; static final int DEFAULT_RETRIES_CONFIG; }### Answer:
@Test public void testDefaultKafkaProducerConfig() { Context context = mock(Context.class); Parameters params = ParameterTestUtils.createParameters(); when(context.getParameters()).then((Answer<Parameters>) invocation -> params); MantisKafkaProducerConfig mantisKafkaProducerConfig= new MantisKafkaProducerConfig(context); Map<String, Object> producerProperties = mantisKafkaProducerConfig.getProducerProperties(); assertEquals(ByteArraySerializer.class, producerProperties.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)); assertEquals(ByteArraySerializer.class, producerProperties.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)); assertEquals(Arrays.asList(MantisKafkaProducerConfig.DEFAULT_BOOTSTRAP_SERVERS_CONFIG), producerProperties.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)); assertEquals(Arrays.asList(JmxReporter.class.getName()), producerProperties.get(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG)); }
@Test public void testJobParamOverrides() { Context context = mock(Context.class); Parameters params = ParameterTestUtils.createParameters(KafkaSinkJobParameters.PREFIX + ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); when(context.getParameters()).then((Answer<Parameters>) invocation -> params); MantisKafkaProducerConfig mantisKafkaProducerConfig= new MantisKafkaProducerConfig(context); Map<String, Object> producerProperties = mantisKafkaProducerConfig.getProducerProperties(); assertEquals(StringSerializer.class, producerProperties.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)); assertEquals(ByteArraySerializer.class, producerProperties.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)); }