Expose HW and NewestOffset in metadata

[LaPetiteSouris]

Refer to issue #111 , metadata should include High Watermark and Newest Offset of each partition.

This PR requires liftbridge-io/liftbridge-api#31 and also liftbridge-io/liftbridge-api#38

[tylertreat]

I need to think about the implementation. As is, there is an issue if the FetchMetadata request goes to a broker which is not a participant in the partition. In this case, the broker won't have messages in the log. We probably need to fetch HW and NewestOffset from the partition leader.

[LaPetiteSouris]

I need to think about the implementation. As is, there is an issue if the FetchMetadata request goes to a broker which is not a participant in the partition. In this case, the broker won't have messages in the log. We probably need to fetch HW and NewestOffset from the partition leader.

Indeed. I missed that . Would it create overhead for the leader in such case ? As the FetchMetaData is a very frequent request since each client will refresh metadata cache regularly.

[tylertreat]

Would it create overhead for the leader in such case ? As the FetchMetaData is a very frequent request since each client will refresh metadata cache regularly.

I'm wondering if this should be a flag on the FetchMetadata request since, generally, this information is not commonly needed? For example, to support #54. By default, we wouldn't fetch this information unless explicitly asked for. Another consideration is that FetchMetadata by default gets metadata for all streams. I wonder if we want to require specific streams if requesting the HW and NewestOffset?

[LaPetiteSouris]

Would it create overhead for the leader in such case ? As the FetchMetaData is a very frequent request since each client will refresh metadata cache regularly.

I'm wondering if this should be a flag on the FetchMetadata request since, generally, this information is not commonly needed? For example, to support #54. By default, we wouldn't fetch this information unless explicitly asked for. Another consideration is that FetchMetadata by default gets metadata for all streams. I wonder if we want to require specific streams if requesting the HW and NewestOffset?

I think that HW and NewestOffset should be fetched on-demand only. Also, with that in mind, I think only HW and NewestOffset of the specific stream should be given. I guess we may not want to "give a jungle with a kingkong holding a banana" while being asked for only "banana".

In that case, I am wondering, should it be given in MetaData at all ? Should it be an independent endpoint on the server side ?

[tylertreat]

A separate endpoint might make sense. That way, the client would use existing metadata cache to know to send the request to the partition leader (much like Subscribe). What that endpoint should look like, I'm not sure, i.e. should it just be specific to fetching HW and NewestOffset or more extensible?

[LaPetiteSouris]

A separate endpoint might make sense. That way, the client would use existing metadata cache to know to send the request to the partition leader (much like Subscribe). What that endpoint should look like, I'm not sure, i.e. should it just be specific to fetching HW and NewestOffset or more extensible?

Sorry for the delays, the past few weeks I was going off-line.

For the specifications of the endpoint, indeed I need to take a look into with some thoughts. Will come back with the API endpoint suggestion soon.

[Jmgr]

Also interested in that feature. Happy to give a hand if needed.

[LaPetiteSouris]

I retake the subject lately and I want to share some propositions:

1. A new api can be created under metadata.go

func (m *metadataAPI) FetchLeaderHW(streams []string) *client.FetchHWandOffsetResponse { }

2. There are 2 strategies:

Strategy A:

func (m *metadataAPI) FetchLeaderHW(streams []string) *client.FetchHWandOffsetResponse { 
	if !m.isLeader() {
		return errors.New(...)
	 }
	// return HW and offset
}

This means by default, it is the client implementation that makes sure the FetchHighestWatermarkandOffset should be sent to leader only. This approach keeps implementation clear, simple and robust.

Strategy B:

func (m *metadataAPI) FetchLeaderHW(streams []string) *client.FetchHWandOffsetResponse { 
	if !m.isLeader() {
		// Propagate a request to the leader to fetch HW and offset
                // return results

	 }
	// return HW and offset
}

This approach ensures that the FetchHighestWatermarkandOffset can arrive to whatever broker , not necessarily the leader. It's the receiving broker that goes and find the information itself. For now I have not thought about how one broker and propagate a request to leader to fetch such information. This approach is more convenient, but implies that we do not keep it "stupid, simple"

What do you think ?

[tylertreat]

@LaPetiteSouris I think you might be confusing the concept of metadata leader with partition/ISR leader. m.IsLeader() indicates if the server is currently the metadata leader, which is simply the Raft leader responsible for control plane operations that go through Raft (create/delete streams, change ISR, etc.).

However, partitions also have a leader who is responsible for appending messages to the commit log from which followers replicate. This is known as the data plane. It is the partition leader who knows what the current HW and latest offset is for a partition.

So with that in mind, the request would need to get forwarded to the broker that is the partition leader. I might have a similar need for the ability to send requests to a partition leader for my implementation of #214, so I think we should wait on that to see if we can reuse the solution here. I'm hoping to have a first pass of that by the end of this week.

[LaPetiteSouris]

@LaPetiteSouris I think you might be confusing the concept of metadata leader with partition/ISR leader. m.IsLeader() indicates if the server is currently the metadata leader, which is simply the Raft leader responsible for control plane operations that go through Raft (create/delete streams, change ISR, etc.).

However, partitions also have a leader who is responsible for appending messages to the commit log from which followers replicate. This is known as the data plane. It is the partition leader who knows what the current HW and latest offset is for a partition.

So with that in mind, the request would need to get forwarded to the broker that is the partition leader. I might have a similar need for the ability to send requests to a partition leader for my implementation of #214, so I think we should wait on that to see if we can reuse the solution here. I'm hoping to have a first pass of that by the end of this week.

Thanks for the clarification. It's true, I was confused.

If I resume it shortly, the client should be "stateful" about who is the leader of the requested partition. And the API side should check and raise error if the request is sent to a broker which is not the partition leader.

With that in mind, the API may look like

func (m *metadataAPI) FetchLeaderHW(streams string, id int32) *client.FetchHWandOffsetResponse { 
	partition := m.GetPartition(stream, id)
        if !partition.IsLeader() {
        // error
       }
}

And it looks like the metadata response already indicates the leader for a given partition id in createMetadataResponse .

[tylertreat]

I think it makes sense to require the request be made to the partition leader from the client, at least for now. In the future, we could add the ability to propagate the request server-side if we decide we want to add it later.

[LaPetiteSouris]

I think it makes sense to require the request be made to the partition leader from the client, at least for now. In the future, we could add the ability to propagate the request server-side if we decide we want to add it later.

Agree, it's simple that way.

When you said you might need the feature to send request to partition leader, which is required for #214 , do you mean the implementation may share common code base on the client side ? Or you may implement the changes on server side as well ?

Anyway, please tag me in if those changes are available so that I can rebase this PR accordingly.

[tylertreat]

When you said you might need the feature to send request to partition leader, which is required for #214 , do you mean the implementation may share common code base on the client side ? Or you may implement the changes on server side as well ?

Initially I was thinking I would implement a server-side mechanism for propagating a request to a partition leader, but I've decided to go a similar route putting the responsibility on the client for the time being.

[tylertreat]

@LaPetiteSouris I will be implementing client-side code for making an RPC to the partition leader, so this should be reusable.

[LaPetiteSouris]

@LaPetiteSouris I will be implementing client-side code for making an RPC to the partition leader, so this should be reusable.

Great news. In that case I am wondering would it be OK if I start implement the API function to handle Highest Watermark and Newest Offset request now . All shall I wait to rebase from you PR ?

[tylertreat]

You should be able to implement the server-side API. I'm hoping to have my client code available by the end of this week.

[LaPetiteSouris]

You should be able to implement the server-side API. I'm hoping to have my client code available by the end of this week.

The server side implementation is committed to the PR.

It requires for now liftbridge-io/liftbridge-api#38

Moreover, due to the fact that the client side is not implemented, the new rpc endpoint's test cases are not yet available.