feat: Adds lag reporting and API for use in lag aware routing as desc by AlanConfluent · Pull Request #1 · AlanConfluent/ksql

AlanConfluent · 2020-01-23T19:14:56Z

Description

What behavior do you want to change, why, how does your patch achieve the changes?

Testing done

Describe the testing strategy. Unit and integration tests are expected for any behavior changes.

Reviewer checklist

Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
Ensure relevant issues are linked (description should include text like "Fixes #")

vpapavas · 2020-01-24T00:03:38Z

+    final long nowMs = clock.millis();
+    final HostInfoEntity hostInfoEntity = lagReportingRequest.getHostInfo();
+    final HostInfo hostInfo = new HostInfo(hostInfoEntity.getHost(), hostInfoEntity.getPort());
+    receivedLagInfo.remove(hostInfo);


Why do you remove the entry here? All you want to do is update the offsets and lag information if the current request has newer information. Keep in mind, requests may arrive out of order

I remove the entry so as to produce one consistent view of the data, per host. I could check the timestamps and ignore if it's an older request, if that's what you mean. I'll do that.

If you don't remove data for this host, then over time, it will be littered with state store that no longer exist or partitions that the host no longer supports. Maybe this isn't a huge deal, but why not be consistent.

vinothchandar

Made a high level pass. Can do a deeper pass once again sometime once I settled there

vinothchandar · 2020-01-24T05:30:27Z

+ * This represents a unique store in the system and the basis for getting lag information from
+ * KafkaStreams, exposed by the QueryMetadata.
+ */
+public final class LagInfoKey {


wonder if this class should reside somewhere close to the agent itself.. ksql-engine is more for the core sql objects/constructs I think?

That makes sense, though we're in need of some way to refer to a state store more generally. At request time, we'll need to pick through persistent query state in order to create this, which isn't easy since it's kind of hidden. It might make sense to add some method getStateStoreId which returns this kind of object from one of the persistent query data structures (and has access to internal state like state store name), and generalizes it a bit instead of making this key lag related. What do you think? Also, which persistent query data structure would own that StateStoreId?

I made a class QueryStateStoreId as a rest entity since it's used as part of the API and has jackson annotations. Tell me if you have another recommendation here.

vinothchandar · 2020-01-24T05:33:32Z

+ * KafkaStreams, exposed by the QueryMetadata.
+ */
+public final class LagInfoKey {
+  private static final String SEPARATOR = "$";


there is a KLIP-16 on k$ dynamic views.. might want to check that once to see if the $ is still a safe separator

vinothchandar · 2020-01-24T06:28:59Z

+  /**
+   * A listener for heartbeat related events.
+   */
+  public interface HeartbeatListener {


HostStatusListener?

nit: we use Node in the KsLocator code.. and use Host here? might be good to use one?

I guess I was thinking there might be other events, but maybe that was overthinking it. Changed to HostStatusListener.

That's not a bad idea to factor out a single host entity, though I think Vicky is adding status to Node, and I'm planning on adding Lag, so Node will represent the joining of all of this data. Maybe it will make sense to revisit that idea after the routing changes are made.

vinothchandar · 2020-01-24T06:31:16Z

+    ).define(
+        KSQL_LAG_REPORTING_SEND_INTERVAL_MS_CONFIG,
+        Type.LONG,
+        1000L,


can we make this something like 3 or 5 seconds. I know it sounds conservative.. but since this is RPC on kafka, best to err on that side

Sure. Choosing 5 seconds

vinothchandar · 2020-01-24T06:33:24Z

+
+      final Map<String, Map<Integer, LagInfo>> storeToPartitionToLagMap = currentQueries.stream()
+          .filter(Objects::nonNull)
+          .flatMap(pmd -> pmd.getStoreToPartitionToLagMap().entrySet().stream()


could we just access the KafkaStreams instance here and fetch the lag map on this class, instead of QueryMetadata? Technically, the lag is not metadata about the query itself and it might be best to consolidate all the lag related code, to the agent

Sure. Moved the code that fetches the map to this class.

vinothchandar · 2020-01-24T06:39:26Z

+   * and debug resources.
+   */
+  public Map<String, Map<String, Map<Integer, LagInfoEntity>>> listAllLags() {
+      return receivedLagInfo.entrySet().stream()


this block of code, I find it hard to read.. Do you think we can just have a class TablePartitionLagInfo which wraps a queryId, storeName, partition, lagInfo and simply manage a list/map tht is just one level? I went with such a nested map because thats how Streams code was.. but I think we can add an abstraction here, to make this easier?

To make it a single level which would be usable by Jackson (which this is since it's exposed in an endpoint) would require "stringifying" the host, storename, partition as a single key and would make it harder to pick apart
in tests. I'll instead use Jackson-friendly types rather than Strings if that would make it easier to read.

I'm fine to rewrite this as a normal block without using so many streams. I went a little overboard :-)

vinothchandar · 2020-01-24T06:44:24Z

+    Map<LagInfoKey, Map<Integer, LagInfo>> storeToPartitionToLagMap = null;
+    try {
+      storeToPartitionToLagMap = kafkaStreams.allLocalStorePartitionLags().entrySet().stream()
+          .map(e -> Pair.of(LagInfoKey.of(getQueryApplicationId(), e.getKey()), e.getValue()))


(I think you are handling this already). Note to self: When a new query is scheduled, it's possible that the sending node has the store created for it already i.e the aware of the query while the receiving node still may not have seen the queryID being scheduled on it.. So if we expect the receiving node to do any checks based on the queryApplicationId(), it may fail in a small window

vinothchandar · 2020-01-24T06:45:55Z

+
+  @Override
+  public void onHostStatusUpdated(final Map<String, HostStatusEntity> hostsStatusMap) {
+    aliveHosts.clear();


do we need to synchronize between this update and the send lag iterating over the map to send lag to other nodes?

Yes, good call. There's technically a race as it is here.

I changed it to create an immutable set and then updates an AtomicReference.

vinothchandar · 2020-01-24T06:46:56Z

+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+@Path("/lag")


same comment as in vicky's PR.. should we just consolidate all of this into the cluster status resource.. atleast the active/standby info and lag info for the same partitions reported on that endpoint, can be consolidated?

Done. Moved the lag info to that endpoint.

Reporting still belongs in its own resource though, so I left that here.

vinothchandar · 2020-01-24T06:48:17Z

+
+@Immutable
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class LagInfoEntity {


note to self : this class is pretty much same as LagInfo in streams.. but we need all these jackson annotations.. so had to create a new one

There's that and also the fact that the constructor of LagInfo is not public. If you could make those changes to the original object, this could be removed.

…ribed in KLIP-12

vpapavas reviewed Jan 24, 2020

View reviewed changes

vinothchandar reviewed Jan 24, 2020

View reviewed changes

AlanConfluent added 20 commits January 29, 2020 14:00

feat: Adds lag reporting and API for use in lag aware routing as desc…

93276a4

…ribed in KLIP-12

Uses new heartbeat listener

a3a29a0

Fixes style issues

ec54539

Remove prev warning removals

a6cb987

Adds lagagent as a listener

e93ac04

Introduces simple LagCache

82090fe

Some comments and simple changes

356ad50

Only allows LagReportingAgent if HeartbeatAgent is enabled

1e836b4

Adds Lag Reporting Functional test

ccb70a9

Style fixes

81f22e1

Adds LagInfoKey to keep track of lags

2ea201e

Style again

459360e

Refactors to key by host

4a4626c

Feedback

e1b7d24

Feedback round 1

77d94fe

More feedback about consolidating cluster status

1260354

Renamed HeartbeatListener to HostStatusListener

1781831

Adds empty set of live hosts to begin with

65b21a8

Moves the null check so it's after getLocalLagMap

e9e7d3c

Feedback from Almog

f35a7bc

AlanConfluent force-pushed the lag_reporting branch from d3e4f95 to f35a7bc Compare January 29, 2020 23:22

AlanConfluent added 2 commits January 29, 2020 17:00

Adds lags to ClusterStatusResponse.equals

ee30b2a

Changes separator to # rather than $

6201d14

Conversation

AlanConfluent commented Jan 23, 2020

Description

Testing done

Reviewer checklist

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vinothchandar left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AlanConfluent Jan 24, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

AlanConfluent Jan 24, 2020 •

edited

Loading