@Test public void populateKafkaPropertiesFromHadoop() { conf.set(BOOTSTRAP_SERVERS.key(), "broker"); conf.set(CLIENT_ID.key(), "client"); conf.set(ACKS.key(), "acknowledgements"); conf.set(RETRIES.key(), "1"); conf.set(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION.key(), "2"); conf.set(BATCH_SIZE.key(), "3"); conf.set(LINGER_MS.key(), "4"); conf.set(BUFFER_MEMORY.key(), "5"); Properties props = kafkaProperties(conf); assertThat(props.get("bootstrap.servers")).isEqualTo("broker"); assertThat(props.get("acks")).isEqualTo("acknowledgements"); assertThat(props.get("retries")).isEqualTo(1); assertThat(props.get("max.in.flight.requests.per.connection")).isEqualTo(2); assertThat(props.get("batch.size")).isEqualTo(3); assertThat(props.get("linger.ms")).isEqualTo(4L); assertThat(props.get("buffer.memory")).isEqualTo(5L); assertThat(props.get("key.serializer")).isEqualTo(LongSerializer.class.getName()); assertThat(props.get("value.serializer")).isEqualTo(ByteArraySerializer.class.getName()); }
@VisibleForTesting static Properties kafkaProperties(Configuration conf) { Properties props = new Properties(); props.put(BOOTSTRAP_SERVERS.unprefixedKey(), checkNotNull(stringProperty(conf, BOOTSTRAP_SERVERS), "Property " + BOOTSTRAP_SERVERS + " is not set")); props.put(CLIENT_ID.unprefixedKey(), checkNotNull(stringProperty(conf, CLIENT_ID), "Property " + CLIENT_ID + " is not set")); props.put(ACKS.unprefixedKey(), stringProperty(conf, ACKS)); props.put(RETRIES.unprefixedKey(), intProperty(conf, RETRIES)); props.put(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION.unprefixedKey(), intProperty(conf, MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)); props.put(BATCH_SIZE.unprefixedKey(), intProperty(conf, BATCH_SIZE)); props.put(LINGER_MS.unprefixedKey(), longProperty(conf, LINGER_MS)); props.put(BUFFER_MEMORY.unprefixedKey(), longProperty(conf, BUFFER_MEMORY)); props.put("key.serializer", "org.apache.kafka.common.serialization.LongSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); return props; }
KafkaMessageSender { @VisibleForTesting static Properties kafkaProperties(Configuration conf) { Properties props = new Properties(); props.put(BOOTSTRAP_SERVERS.unprefixedKey(), checkNotNull(stringProperty(conf, BOOTSTRAP_SERVERS), "Property " + BOOTSTRAP_SERVERS + " is not set")); props.put(CLIENT_ID.unprefixedKey(), checkNotNull(stringProperty(conf, CLIENT_ID), "Property " + CLIENT_ID + " is not set")); props.put(ACKS.unprefixedKey(), stringProperty(conf, ACKS)); props.put(RETRIES.unprefixedKey(), intProperty(conf, RETRIES)); props.put(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION.unprefixedKey(), intProperty(conf, MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)); props.put(BATCH_SIZE.unprefixedKey(), intProperty(conf, BATCH_SIZE)); props.put(LINGER_MS.unprefixedKey(), longProperty(conf, LINGER_MS)); props.put(BUFFER_MEMORY.unprefixedKey(), longProperty(conf, BUFFER_MEMORY)); props.put("key.serializer", "org.apache.kafka.common.serialization.LongSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); return props; } }
KafkaMessageSender { @VisibleForTesting static Properties kafkaProperties(Configuration conf) { Properties props = new Properties(); props.put(BOOTSTRAP_SERVERS.unprefixedKey(), checkNotNull(stringProperty(conf, BOOTSTRAP_SERVERS), "Property " + BOOTSTRAP_SERVERS + " is not set")); props.put(CLIENT_ID.unprefixedKey(), checkNotNull(stringProperty(conf, CLIENT_ID), "Property " + CLIENT_ID + " is not set")); props.put(ACKS.unprefixedKey(), stringProperty(conf, ACKS)); props.put(RETRIES.unprefixedKey(), intProperty(conf, RETRIES)); props.put(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION.unprefixedKey(), intProperty(conf, MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)); props.put(BATCH_SIZE.unprefixedKey(), intProperty(conf, BATCH_SIZE)); props.put(LINGER_MS.unprefixedKey(), longProperty(conf, LINGER_MS)); props.put(BUFFER_MEMORY.unprefixedKey(), longProperty(conf, BUFFER_MEMORY)); props.put("key.serializer", "org.apache.kafka.common.serialization.LongSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); return props; } KafkaMessageSender(Configuration conf); @VisibleForTesting  KafkaMessageSender(String topic, KafkaProducer<Long, byte[]> producer); }
KafkaMessageSender { @VisibleForTesting static Properties kafkaProperties(Configuration conf) { Properties props = new Properties(); props.put(BOOTSTRAP_SERVERS.unprefixedKey(), checkNotNull(stringProperty(conf, BOOTSTRAP_SERVERS), "Property " + BOOTSTRAP_SERVERS + " is not set")); props.put(CLIENT_ID.unprefixedKey(), checkNotNull(stringProperty(conf, CLIENT_ID), "Property " + CLIENT_ID + " is not set")); props.put(ACKS.unprefixedKey(), stringProperty(conf, ACKS)); props.put(RETRIES.unprefixedKey(), intProperty(conf, RETRIES)); props.put(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION.unprefixedKey(), intProperty(conf, MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)); props.put(BATCH_SIZE.unprefixedKey(), intProperty(conf, BATCH_SIZE)); props.put(LINGER_MS.unprefixedKey(), longProperty(conf, LINGER_MS)); props.put(BUFFER_MEMORY.unprefixedKey(), longProperty(conf, BUFFER_MEMORY)); props.put("key.serializer", "org.apache.kafka.common.serialization.LongSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); return props; } KafkaMessageSender(Configuration conf); @VisibleForTesting  KafkaMessageSender(String topic, KafkaProducer<Long, byte[]> producer); void send(KafkaMessage kafkaMessage); }
KafkaMessageSender { @VisibleForTesting static Properties kafkaProperties(Configuration conf) { Properties props = new Properties(); props.put(BOOTSTRAP_SERVERS.unprefixedKey(), checkNotNull(stringProperty(conf, BOOTSTRAP_SERVERS), "Property " + BOOTSTRAP_SERVERS + " is not set")); props.put(CLIENT_ID.unprefixedKey(), checkNotNull(stringProperty(conf, CLIENT_ID), "Property " + CLIENT_ID + " is not set")); props.put(ACKS.unprefixedKey(), stringProperty(conf, ACKS)); props.put(RETRIES.unprefixedKey(), intProperty(conf, RETRIES)); props.put(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION.unprefixedKey(), intProperty(conf, MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)); props.put(BATCH_SIZE.unprefixedKey(), intProperty(conf, BATCH_SIZE)); props.put(LINGER_MS.unprefixedKey(), longProperty(conf, LINGER_MS)); props.put(BUFFER_MEMORY.unprefixedKey(), longProperty(conf, BUFFER_MEMORY)); props.put("key.serializer", "org.apache.kafka.common.serialization.LongSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); return props; } KafkaMessageSender(Configuration conf); @VisibleForTesting  KafkaMessageSender(String topic, KafkaProducer<Long, byte[]> producer); void send(KafkaMessage kafkaMessage);  }