Request review and merge of Kafka Connector for forklift project #124

AFrieze · 2017-03-06T21:06:05Z

Kafka connector with confluent schema-registry for forklift. Adapts to JMS specification and should be able to replace existing activemq producers and consumer in the majority of scenarios.

Master

Feature/kafka connector

kaiserleib · 2017-03-06T21:35:11Z

Failing in CI, just FYI

Feature/kafka connector

Removed sofi avro dependency

Kuroshii · 2017-03-08T00:29:02Z

connectors/kafka/src/main/java/forklift/producers/KafkaForkliftProducer.java

+ avroRecord.put("forkliftMsg", message);
+ ProducerRecord record = new ProducerRecord<>(topic, null, avroRecord);
+ try {
+ kafkaProducer.send(record);


We should probably have a callback on all of the kafkaProducer.send calls to at least log errors returned from the broker.

Kuroshii · 2017-03-08T00:40:56Z

connectors/kafka/build.sbt

+version := "0.1"
+
+// target and Xlint cause sbt dist to fail
+javacOptions ++= Seq("-source", "1.8")//, "-target", "1.8", "-Xlint")


You can still have linting on just compile with javacOptions in (Compile, compile).

Kuroshii · 2017-03-08T12:23:26Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ props.put("key.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
+ props.put("value.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
+ props.put("schema.registry.url", schemaRegistries);
+ props.put("specific.avro.reader", false);


I think for most of our cases moving forward we would prefer to read avro as a SpecificRecord for the stronger type checking. Accessing data "as a map" is usually too flexible.

Also, it seems strange to support writing specific records, but not reading specific records.

Kuroshii · 2017-03-08T12:34:30Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+
+ static ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule())
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ KafkaController controller;


Is there a reason to have these two variables package-local? Just for testing?

Yea I could prefer to see these be protected because of general confusion around the default modifier if the access modifier is just for testing

Kuroshii · 2017-03-08T12:43:46Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+
+ @Override
+ public ForkliftProducerI getTopicProducer(String name) {
+ return new KafkaForkliftProducer(name, this.kafkaProducer);


What if the kafkaProducer is null when this is called (i.e. start has not been called yet, or is in the process of being called)? Maybe an IllegalStateException would be appropriate.

But that still seems a little weird because you can somehow get a consumer without starting the connector but you can't get a producer without starting the connector.

Kuroshii · 2017-03-08T13:08:55Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ this.controller.start();
+ }
+ }
+ return new KafkaTopicConsumer(name, controller, messageStream);


Design nit: I think it would make more sense to have the controller manage the details of creating a consumer. That way you can put the logic for setting up a consumer to use a controller in the controller rather than having the consumer manage state to keep track of whether the controller is actually completely set up for it. This also has the minor benefit of avoiding exposing the MessageStream object to the connector if it doesn't need to be exposed.

Kuroshii · 2017-03-08T13:37:18Z

connectors/kafka/src/main/java/forklift/producers/KafkaForkliftProducer.java

+ try {
+ kafkaProducer.send(record);
+ } catch (SerializationException e) {
+ throw new ProducerException("Error creating Kafka Message", e);


Copy-pasted exception? I don't think this message makes sense here

Kuroshii · 2017-03-08T13:37:53Z

connectors/kafka/src/main/java/forklift/producers/KafkaForkliftProducer.java

+ }
+
+ @Override
+ public String send(Object message) throws ProducerException {


Minor: this method could use a little style clean up

Kuroshii · 2017-03-08T13:44:43Z

connectors/kafka/src/main/java/forklift/producers/KafkaForkliftProducer.java

+
+ String mapSchema = "{\"type\":\"record\"," +
+ "\"name\":\"ForkliftMapMessage\"," +
+ "\"fields\":[{\"name\":\"forkliftMapMsg\",\"type\":\"string\"}]}";


Avro has a map type; can we use that instead of a string?

I didn't think it was supported by our serializer

What do you mean? It's definitely in the avro spec, at least...

Kuroshii · 2017-03-08T13:55:21Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ GenericRecord genericRecord = (GenericRecord)record.value();
+ Object value = genericRecord.get("forkliftMapMsg");
+ value = value == null ? genericRecord.get("forkliftMsg") : value;
+ value = value == null ? genericRecord.get("forkliftJsonMsg") : value;


Shouldn't we deserialize these values from our message so that our message's value appears to be what was sent? I don't see anything deserializing the string that was constructed in the producer for maps, for example.

Also, with removing the JMS stuff, this is going to need to find a new home...

The string is parsed to the appropriate type based on the annotated Message object. Look at the inject method in Consumer.java

I see, thanks.

Kuroshii · 2017-03-08T13:58:40Z

connectors/kafka/src/main/java/forklift/connectors/KafkaController.java

+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetData = this.acknowlegmentHandler.flushAcknowledged();
+ if (offsetData.size() > 0) {
+ String offsetDescription =


I would rather not do this string construction all the time if it only matters for debug logs.

It would be cool if we could just throw lazy before this and call it good, but alas.

sagersmith8 · 2017-03-09T16:22:20Z

connectors/kafka/build.sbt

+ "com.fasterxml.jackson.datatype" % "jackson-datatype-jsr310" % "2.7.3",
+ "org.apache.kafka" % "kafka-clients" % "0.10.1.1-cp1",
+ "io.confluent" % "kafka-avro-serializer" % "3.1.1"
+


Extra newline

sagersmith8 · 2017-03-09T16:23:26Z

connectors/kafka/build.sbt

+lazy val testDependencies = Seq(
+ "commons-io" % "commons-io" % "2.4" ,
+ "com.novocode" % "junit-interface" % "0.11",
+ "org.mockito" % "mockito-core" % "1.9.5"


Weird extra spaces should be:

"org.mockito" % "mockito-core" % "1.9.5"

sagersmith8 · 2017-03-09T16:24:13Z

connectors/kafka/build.sbt

+ if (isSnapshot.value)
+ Some("snapshots" at nexus + "content/repositories/snapshots")
+ else
+ Some("releases" at nexus + "service/local/staging/deploy/maven2")


Extra space, should be:

Some("releases" at nexus + "service/local/staging/deploy/maven2")

sagersmith8 · 2017-03-09T16:25:01Z

connectors/kafka/build.sbt

+
+publishTo := {
+ val nexus = "https://oss.sonatype.org/"
+ if (isSnapshot.value)


Best not to use single line if

sagersmith8 · 2017-03-09T16:25:12Z

connectors/kafka/build.sbt

+ val nexus = "https://oss.sonatype.org/"
+ if (isSnapshot.value)
+ Some("snapshots" at nexus + "content/repositories/snapshots")
+ else


Best not to use single line else

sagersmith8 · 2017-03-09T16:29:16Z

connectors/kafka/src/main/java/forklift/connectors/AcknowledgedRecordHandler.java

+ * Acknowledged records are those that have started processing but have not yet been committed to the Kafka Broker. This class is threadsafe.
+ */
+public class AcknowledgedRecordHandler {
+ private static final Logger log = LoggerFactory.getLogger(AcknowledgedRecordHandler.class);


Should be named LOGGER per java convention on static final varaibles

sagersmith8 · 2017-03-09T16:33:14Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ private KafkaProducer<?, ?> kafkaProducer;
+ private MessageStream messageStream = new MessageStream();
+
+ static ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule())


This could be final

sagersmith8 · 2017-03-09T16:34:28Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+
+ @Override
+ public void start() throws ConnectorException {
+


Extra newline

sagersmith8 · 2017-03-09T16:34:53Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ this.controller.start();
+ }
+
+ private KafkaController createController() {


Missing javadocs

sagersmith8 · 2017-03-09T16:35:21Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ props.put("value.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
+ props.put("schema.registry.url", schemaRegistries);
+ props.put("specific.avro.reader", false);
+ //props.put("auto.offset.reset", "earliest");


If you don't need this you should remove

sagersmith8 · 2017-03-09T16:37:09Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ } catch (InterruptedException e) {
+ log.error("KafkaConnector interrupted while stopping");
+ }
+ this.kafkaProducer.close();


Probably wrap this in a finally is my guess`

sagersmith8 · 2017-03-09T16:39:02Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ }
+
+ @Override
+ public ForkliftMessage jmsToForklift(Message m) {


No single character variables please, it makes debugging hard

sagersmith8 · 2017-03-09T16:40:07Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ msg.setMsg(value.toString());
+ }
+ } else {
+ ObjectMapper mapper = new ObjectMapper();


Why don't you use the one that you declared at the top?

sagersmith8 · 2017-03-09T16:42:17Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ GenericRecord genericRecord = (GenericRecord)record.value();
+ Object value = genericRecord.get("forkliftMapMsg");
+ value = value == null ? genericRecord.get("forkliftMsg") : value;
+ value = value == null ? genericRecord.get("forkliftJsonMsg") : value;


Doesn't this line just overwrite the last line? This seems like you are trying to do a weird cascading check, but it won't actually tell you what went wrong if that is the case. Why don't you just use if checks?

sagersmith8 · 2017-03-09T16:43:37Z

connectors/kafka/src/main/java/forklift/connectors/KafkaConnector.java

+ try {
+ //inefficient as we map from object to json, then back to the object. This approach works
+ //without any changes to the forklift core libraries however.
+ msg.setMsg(mapper.writeValueAsString(record.value()));


again why don't you use the ObjectMapper declared at the top>

AFrieze · 2017-03-09T16:45:45Z

All - Thank you for the feedback. We have decided to keep the kafka functionality in a project specific to Kafka for now to allow us to make more changes to the core and move away from the JMS dependencies. We may have a larger PR coming in for that in the future if that's the way forklift wants to go. All the feedback is very much appreciated.

AFrieze and others added 7 commits March 2, 2017 15:15

Merge pull request #2 from dcshock/master

cc797d6

Master

Initial implementation of Kafka Connector

b157332

added unit tests

7f405a0

Fixed some synchronization bugs, additional work on integration tests

0a6df72

Added readme

87808cd

updated server readme

ce26f29

Merge pull request #3 from AFrieze/feature/kafka-connector

688d4f3

Feature/kafka connector

Andrew Frieze and others added 5 commits March 7, 2017 08:33

added more to README

e775481

Added confluent repo to built.sbt

7706f47

Merge pull request #5 from AFrieze/feature/kafka-connector

2bc6137

Feature/kafka connector

Removed sofi avro dependency

2d15b04

Merge pull request #6 from AFrieze/feature/kafka-connector

95e49d2

Removed sofi avro dependency

Kuroshii reviewed Mar 8, 2017

View reviewed changes

sagersmith8 reviewed Mar 9, 2017

View reviewed changes

connectors/kafka/build.sbt

publishTo := {

val nexus = "https://oss.sonatype.org/"

if (isSnapshot.value)

Copy link

sagersmith8 Mar 9, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Best not to use single line if

sagersmith8 reviewed Mar 9, 2017

View reviewed changes

AFrieze closed this Mar 9, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Request review and merge of Kafka Connector for forklift project #124

Request review and merge of Kafka Connector for forklift project #124

AFrieze commented Mar 6, 2017

kaiserleib commented Mar 6, 2017

Kuroshii Mar 8, 2017

Kuroshii Mar 8, 2017

Kuroshii Mar 8, 2017

Kuroshii Mar 8, 2017

sagersmith8 Mar 9, 2017

Kuroshii Mar 8, 2017 •

edited

Loading

Kuroshii Mar 8, 2017

Kuroshii Mar 8, 2017

Kuroshii Mar 8, 2017

Kuroshii Mar 8, 2017

AFrieze Mar 8, 2017

Kuroshii Mar 8, 2017

Kuroshii Mar 8, 2017

AFrieze Mar 8, 2017

Kuroshii Mar 8, 2017

Kuroshii Mar 8, 2017

AFrieze Mar 8, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

sagersmith8 Mar 9, 2017

AFrieze commented Mar 9, 2017

Request review and merge of Kafka Connector for forklift project #124

Request review and merge of Kafka Connector for forklift project #124

Conversation

AFrieze commented Mar 6, 2017

kaiserleib commented Mar 6, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Kuroshii Mar 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AFrieze commented Mar 9, 2017

Kuroshii Mar 8, 2017 •

edited

Loading