KAFKA-9726: Add IdentityReplicationPolicy for MM2

[mdedetrich]

I am taking over the PR of #9395 which has been rebased to work with the latest trunk of Kafka. As stated in the original PR, it adds a IdentityReplicationPolicy for MirrorMaker2 (known as LegacyReplicationPolicy) which imitates the behavior of MirrorMaker1 (i.e. it doesn't rename the the topics apart from heartbeat). The behavior of this PR is exactly the same as the stated original.

Here are some additional notes/changes that are different from the original PR.

• An additional constructor for the startClusters() has been added which allows you to supply test specific MirrorMaker configs. This allows the IdentityReplicationPolicy tests to use as much more code from the base MirrorConnectorsIntegrationBaseTest, heavily reducing boilerplate when compared to the original PR. Although It can be argued that you can just do .put directly on the mm2Props HashMap, this method is cleaner and more importantly better handles the case where you need to override an existing key (since super needs to be called last in startClusters it will override any existing keys which you may have configured in your own test)
• Certain methods/fields of MirrorConnectorsIntegrationBaseTest have been changed from private to protected so that IdentityReplicationPolicy specific tests can access them.

@ryannedolan You reviewed the original PR and in that PR you stated that a KIP is needed since we are adding a public method to the ReplicationPolicy interface, is this still necessary? You also had issues with the canTrackSource method name, do you have a better alternative?

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

EDIT: Renamed LegacyReplicationPolicy to IdentityReplicationPolicy

[mdedetrich]

There seems to be spurious test failures that is unrelated to the PR, let me know if I need to do something about it (i.e. retrigger the CI?)

[mimaison]

Thanks for the PR!

As this is adding a new method to a public interface (ReplicationPolicy), I'm afraid this will require a KIP.

Also what about naming the new policy IdentityReplicationPolicy? Legacy can have a negative connotation and I've seen such a policy referred as "Identity" in a few places already.

[ryannedolan]

Thanks for picking this up. I have a similar effort here: #10652 (just posted).

If we need to modify the interface, we'll definitely need a KIP. My PR doesn't change the interface, so I believe it's possible to avoid the KIP. That said, my PR is currently not passing, so maybe you are ahead of me :)

wrt Legacy vs Identity, I am fine either way. I'm happy to touch-up KIP-382 to avoid confusion if we go with Identity.

[mdedetrich]

@mimaison I have just renamed LegacyReplicationPolicy to IdentityReplicationPolicy and forced pushed the branch.

@ryannedolan I will have a look at #10652 early next week to see if I can avoid adding the method to the public interface otherwise I will create a KIP as it appears to be required. Are you guys happy with the canTrackSource name should it be needed?

[mdedetrich]

KIP has been created at https://cwiki.apache.org/confluence/display/KAFKA/KIP-737%3A+Add+canTrackSource+to+ReplicationPolicy and a new thread has been started in the kafka-dev mailing list for this topic (let me know if anything more needs to be done).

[ryannedolan]

I still don't think an API change is required here. I was able to get your integration tests passing with this version of IdentityReplicationPolicy: https://github.com/apache/kafka/pull/10652/files#diff-79a09517576a35906123533490ed39c0e1a9416878e284d7b71f5f4c53eeca29R31

I just had to add this extra check to MirrorSourceConnector:

https://github.com/apache/kafka/pull/10652/files#diff-b7d6db5cc72b500fab6c628f42376198eaecfd6258069bbff0e2ec98ee9e9427R497

which I think is probably harmless. Thoughts?

[mdedetrich]

As discussed in the mailing list, the other PR is more suitable since it doesn't require a KIP, I will close this PR (it can be reopened later if something else changes)