[ https://issues.apache.org/jira/browse/KAFKA-3914?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15355415#comment-15355415 ]
ASF GitHub Bot commented on KAFKA-3914: --------------------------------------- GitHub user dguy opened a pull request: https://github.com/apache/kafka/pull/1570 KAFKA-3914: Global discovery of state stores @guozhangwang @enothereska please take a look. A few things that need to be clarified 1. I've added `StreamsConfig.USER_ENDPOINT_CONFIG`, but should we have separate configs for host and port or is this one config ok? 2. `HostState` in the KIP has a `byte[]` field - not sure why and what it would be populated with 3. The API calls in the KIP all return `Map<HostState, Set<TaskMetadata>>`, however i don't see why the `Set` is required so I've changed it to `Map<HostState, TaskMetadata>` You can merge this pull request into a Git repository by running: $ git pull https://github.com/dguy/kafka kafka-3914 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/kafka/pull/1570.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #1570 ---- commit f22ca0e680063d42240d8ceebe140ecd9e3fab48 Author: Damian Guy <damian....@gmail.com> Date: 2016-06-29T07:40:11Z Connect stateStores to source topics. Add user endpoint config to StreamsConfig. Collect Map<HostState, Set<TopicPartition>> in StreamPartitionAssignor API for retrieving Map<HostState, TaskMetadata> ---- > Global discovery of state stores > -------------------------------- > > Key: KAFKA-3914 > URL: https://issues.apache.org/jira/browse/KAFKA-3914 > Project: Kafka > Issue Type: Sub-task > Components: streams > Affects Versions: 0.10.1.0 > Reporter: Damian Guy > Assignee: Damian Guy > Fix For: 0.10.1.0 > > > - Update group membership data to include discovery endpoints > - Enable discovery > We need to attach some host and port information to > {{org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo}} > This can then be used in > {{org.apache.kafka.streams.processor.internals.StreamPartitionAssignor}} to > build a {{Map<HostState, Set<TopicPartition>>}} that should be added to > {{org.apache.kafka.streams.processor.internals.assignment.AssignmentInfo}} > during the {{assign}} call. > When {{StreamPartitionAssignor.onAssignment}} is called we should hold on to > a copy of the {{Map<HostState, Set<TopicPartition>>}} for use by the > discovery methods > To enable discovery, implement the below methods on {{KafkaStreams}} > {code} > /** > * @return metadata about all tasks > */ > Map<HostState, Set<TaskMetadata>> getAllTasks(); > /** > * @param storeName requested store name > * @return metadata about all tasks that include > * storeName in this KStreams instance > */ > Map<HostState, Set<TaskMetadata>> getAllTasksWithStore(String storeName); > /** > * @param key requested key > * @param storeName requested store name > * @return metadata about all tasks that include > * storeName and key in this KStreams instance > */ > <K> Map<HostState, Set<TaskMetadata>> getAllTasksWithKey(String storeName, K > key); > {code} -- This message was sent by Atlassian JIRA (v6.3.4#6332)