[STORM-1015] Allow Kafka offsets to be saved using Kafka's consumer offset management api

[ooasis]

Not sure when it will be reviewed. So I chose to implement it based on master branch. Could be ported to other branches if needed.

-thanks

[rmkellogg]

A few minor nits:

When providing javadoc on variables/methods, be sure to use the following syntax:

/**
* Comment text
**/
private String variableName;

Instead of the following:

// offset state information storage. validate options are storm and kafka
private String variableName;

The first variation will provide fly-over-help within IDE and generated Javadoc files.

Secondly:
Logger instance variables should be private.

Thirdly:
Suggest instance variables on PartitionManager, StaticCoordinator ZkDataStore be made private.

[ooasis]

Thanks for the advices. Will get them fixed as soon as I get some time.

[sweetest]

how will this change affect the format of spout's offset message in zookeeper?

[ooasis]

I assume you are referring to the json format of the spout "state" stored in ZK? The "state" will be saved using the exact same json format in Kafka's internal topic as the "metadata" associated with the offsets.

[ooasis]

Fixed coding styles suggested by rmkellogg.

[erikdw]

please expand this * import to explicitly include the classes you depend on

[ooasis]

The fix is pushed

[erikdw]

This seems like it might not be backwards compatible with the existing kafka-spout; i.e., the offsets in ZK are presumably not going to be stored exactly the same as they were before. Is there any plan for supporting migration from the current kafka-spout to this one?

[erikdw]

Furthermore, can you please provide a reference to "Kafka's consumer offset management api" in your description?

[ooasis]

Erik,

I added 2 reference links in the description with more information on "Kafka's consumer offset management api".

As of migration from existing offsets stored in ZK to Kafka, it could be done with a simple migration utility similar to the tools provided by Kafka (https://cwiki.apache.org/confluence/display/KAFKA/System+Tools). Question is where it belongs to in Storm codebase.

-thanks

[erikdw]

minor nit: Zookeeper -> ZooKeeper.

[erikdw]

@hsun-cnnxty : I don't see the reference links in the Description on the PR's Conversation view? Maybe I'm looking in the wrong place? Maybe they should be in comments in the code too?
Yup, was looking in the wrong place, I didn't realize that "the description" meant the JIRA issue's Description!

I'm guessing the following are the links you meant to put:

• https://cwiki.apache.org/confluence/display/KAFKA/Committing+and+fetching+consumer+offsets+in+Kafka
• https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol

I just want it to be very clear that this PR's purpose is to change from the kafka spout's consumer offsets being stored in ZooKeeper, to instead being stored directly in Kafka. We should also be clear about the version of Kafka required for such support (0.8.2+). I know the current version of the storm-kafka pom.xml (as of this change) is already referencing 0.8.2.1, but I feel like it should be called out as an explicit requirement in the commit.

[erikdw]

typo: original

[ooasis]

Erik,

Pushed the change to rename config from "storm" to "zookeeper" (also fixed the typo). As of Kafka dependency, it is a good question. Actually I am not clear on how the "storm-kafka" module is built with Maven. Without adding the particular Kafka dependencies in pom.xml, the local build just won't work for me. Do I suppose to use some external config (such as settings.xml) to make it built locally?

-thanks

[choang]

I recommend making your abstraction at the state store, so you would have:

public interface StateStore {
  public void write(Partition p, long offset);
  public long read(Partition p);
}

public class ZkStateStore implements StateStore {
...
}

public class KafkaStateStore implements StateStore {
...
}

/**
Some topologies read from LATEST after a restart, so only memory state is needed.
*/
public class MemoryStateStore implements StateStore {
...
}

You would not need different PartitionStateManager for different stores. You would just have:

public class PartitionStateManager {
    public PartitionStateManager (..., StateStore store) { ... }
    public void writeState(...) {
        store.write(...);
    }
}

[ooasis]

Hi Chi,

Storm stores more than just offset/partition data in the "state", would it be necessary to declare?

public interface StateStore {
    public void write(Partition p, Map<Object, Object> state);
    public Map<Object, Object> state read(Partition p);
}

-thanks

[choang]

I would make state more concrete, but I suppose your approach is fine.

[erikdw]

@hsun-cnnxty : can you please make your statement a bit more concrete? i.e., what other info is stored in a given kafka topic's consumer state? (other than the offset/partition)

[ooasis]

@choang: the refactored code is pushed

@erikdw: I was referring to the internal Json structure used to store offsets as shown by example below:

{
"broker": {
"host": "kafka.sample.net",
"port": 9092
},
"offset": 4285,
"partition": 1,
"topic": "testTopic",
"topology": {
"id": "fce905ff-25e0 -409e-bc3a-d855f 787d13b",
"name": "Test Topology"
}
}

[choang]

is there a mechanism/section in the release notes to call out this change? changing this var name would break existing code.

[choang]

instead of a factory, you can make the developer declare the StateStore:

public void createTopology() {
    Spout spout = new KafkaSpout(..., new KafkaStateStore(...));
...
}

This should keep the KafkaSpout code much simpler and more explicit, and eliminate the need for a factory.

[choang]

stormConf isn't used so remove. By using spoutConfig, you are bleeding the spout state into the store. I understand it is for convenience, but using explicit would be better, so perhaps you can create a KafkaStoreConfig.

[choang]

looks really good. just had a few nits that shouldn't block. I also don't have a vote, so you'll need to find a proper reviewer :)

[ooasis]

@choang thanks for the code review. Your comment on sharing the kafka store for all partitions is really helpful and I was able to cleanup a lot of unnecessary logic. I still have some code to check in and need to run a few tests. Maybe you can hold on for a while as I don't want to waste your time on the changing code base. I will try to write a response as soon as I get some time. Thanks again.

[ooasis]

@choang In recent changes, I have made it possible to plug in custom store implementations. The custom implementation is given the opportunity to initialize itself by injecting two configuration objects into the constructor which takes the form like

public MyStoreImpl(Map conf, SpoutConfig spoutConfig)

In addition, I feel it is better to keep the implementation specific details, such as zookeeper quorum, inside the store implementation instead of leaving it to external class to build a implementation specific configuration object such as KafkaStoreConfig. What do you think?

Thanks again for the code review.

[choang]

StateStore implementations could be designed like:

public class KafkaStateStore implements StateStore {
    public KafkaStateStore(HostPort kafkaBroker, String consumerId) {  // consumerId could be topic?  I'm don't know enough about the compaction feature
        ...
     }
...
}

public class ZkStateStore implements StateStore {
    public ZkStateStore(HostPort zkConnect, String path) {
        ...
    }
}

// future implementations
public class JdbcStateStore implements StateStore {
    public JdbcStateStore(String url, String stateTable, String user, String password) {
        ...
    }
}

public class MemoryStateStore implements StateStore {
    public MemoryStateStore() {
        ...
    }
}

What I'm trying to illustrate are two factors:

1. go away from StormConf and SpoutConfig because StateStore does not need all properties of either objects.
2. the concrete implements of StateStore do not need to have the same constructor because your goal is not to have a factory and make it completely config driven. You want to make it so the topology developer can decide what StateStore to use.

With the above, it would be pretty easy for someone to implement MemoryStateStore and JdbcStateStore.

[ooasis]

@choang changes are made and latest code from master merged in. Please review whenever you have time. -thanks

[choang]

this looks good enough

[AwesomeJohnR]

Any updates on the status of this merge? This update would be very helpful

[ooasis]

Just did another merge to keep it up to date with master branch. Not sure what the plan is now. I would be happy to make any changes that can help to make the final merge happen.