KAFKA-1215: Rack-Aware replica assignment option #132
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -49,16 +49,22 @@ public PartitionState(int controllerEpoch, int leader, int leaderEpoch, List<Int | |
| this.zkVersion = zkVersion; | ||
| this.replicas = replicas; | ||
| } | ||
| } | ||
|
|
||
| public static final class Broker { | ||
| public final int id; | ||
| public final Map<SecurityProtocol, EndPoint> endPoints; | ||
| public final String rack; | ||
|
|
||
| public Broker(int id, Map<SecurityProtocol, EndPoint> endPoints, String rack) { | ||
| this.id = id; | ||
| this.endPoints = endPoints; | ||
| this.rack = rack; | ||
| } | ||
|
|
||
| @Deprecated | ||
| public Broker(int id, Map<SecurityProtocol, EndPoint> endPoints) { | ||
|
There was a problem hiding this comment. This constructor is only used in tests, does it make sense to keep it? I guess the question is whether the request classes are API. As I understand, they are not, but I would like to get @junrao's take. |
||
| this(id, endPoints, null); | ||
|
There was a problem hiding this comment. Related to my protocol question above. Would defaulting to empty string work? There was a problem hiding this comment. I discussed this in KIP discussion. NULLABLE_STRING was recommended in the discussion. I think it makes sense as rack itself is designed to be nullable (Option[String]). It is legal to define rack as an empty string. There isn't really any null checks in the code as far as I can tell. null just means no rack is defined. |
||
| } | ||
| } | ||
|
|
||
|
|
@@ -91,6 +97,7 @@ public EndPoint(String host, int port) { | |
| // Broker key names | ||
| private static final String BROKER_ID_KEY_NAME = "id"; | ||
| private static final String ENDPOINTS_KEY_NAME = "end_points"; | ||
| private static final String RACK_KEY_NAME = "rack"; | ||
|
|
||
| // EndPoint key names | ||
| private static final String HOST_KEY_NAME = "host"; | ||
|
|
@@ -117,20 +124,20 @@ private static Set<Broker> brokerEndPointsToBrokers(Set<BrokerEndPoint> brokerEn | |
| for (BrokerEndPoint brokerEndPoint : brokerEndPoints) { | ||
| Map<SecurityProtocol, EndPoint> endPoints = Collections.singletonMap(SecurityProtocol.PLAINTEXT, | ||
| new EndPoint(brokerEndPoint.host(), brokerEndPoint.port())); | ||
| brokers.add(new Broker(brokerEndPoint.id(), endPoints, null)); | ||
|
There was a problem hiding this comment. Could use the constructor that doesn't take a rack. There was a problem hiding this comment. @granthenke Given the comment from @junrao that the old constructor should be deprecated, I think it is better to use the new constructor. |
||
| } | ||
| return brokers; | ||
| } | ||
|
|
||
| /** | ||
| * Constructor for version 2. | ||
|
There was a problem hiding this comment. We will still need a separate constructor for v1 of UpdateMetadataRequest since in ControllerChannelManager, we may need to send a v1 request depending on inter.broker.protocol. There was a problem hiding this comment. @junrao This may not be necessary. See my comment in ControllerChannelManager. There was a problem hiding this comment. Note that this constructor is only used in tests. Does it even make sense to keep it? |
||
| */ | ||
| public UpdateMetadataRequest(int controllerId, int controllerEpoch, Map<TopicPartition, | ||
| PartitionState> partitionStates, Set<Broker> liveBrokers) { | ||
| this(2, controllerId, controllerEpoch, partitionStates, liveBrokers); | ||
| } | ||
|
|
||
| public UpdateMetadataRequest(int version, int controllerId, int controllerEpoch, Map<TopicPartition, | ||
|
There was a problem hiding this comment. Is this only public for testing? Would protected or default also work? There was a problem hiding this comment. No, it is used in ControllerChannelManager and has to be public. There was a problem hiding this comment. Is that because we're depending on this constructor for version 1? I know we depend on choosing the right constructor in other request objects to get the right version, but I wonder if it would be better to have explicit static factory methods (e.g. There was a problem hiding this comment. We depend on this constructor to create version 1 and 2 UpdateMetadataRequest, and possibly for future versions as well. There was a problem hiding this comment. Fair enough. I was only wondering if there was a way to keep the version better encapsulated (like all of the other requests). Perhaps at least there should be a check on the version to make sure it is greater than 1? I might even enforce only version 1 and 2 since we'll almost certainly have to touch this code anyway if there is another version bump. There was a problem hiding this comment. I think my preference would probably be to have static factory methods with the versions included in the name. Using constructors is kind of annoying because you have to check the comment to make sure you get the right one. There was a problem hiding this comment. Yeah, we should use more static factories and less overloaded constructors in Kafka. There was a problem hiding this comment. @ijuma @hachikuji Would you mind if I leave this code refactoring of constructors to you guys? |
||
| PartitionState> partitionStates, Set<Broker> liveBrokers) { | ||
| super(new Struct(ProtoUtils.requestSchema(ApiKeys.UPDATE_METADATA_KEY.id, version))); | ||
| struct.set(CONTROLLER_ID_KEY_NAME, controllerId); | ||
|
|
@@ -173,6 +180,9 @@ private UpdateMetadataRequest(int version, int controllerId, int controllerEpoch | |
|
|
||
| } | ||
| brokerData.set(ENDPOINTS_KEY_NAME, endPointsData.toArray()); | ||
| if (version >= 2) { | ||
| brokerData.set(RACK_KEY_NAME, broker.rack); | ||
| } | ||
| } | ||
|
|
||
| brokersData.add(brokerData); | ||
|
|
@@ -226,8 +236,8 @@ public UpdateMetadataRequest(Struct struct) { | |
| int port = brokerData.getInt(PORT_KEY_NAME); | ||
| Map<SecurityProtocol, EndPoint> endPoints = new HashMap<>(1); | ||
| endPoints.put(SecurityProtocol.PLAINTEXT, new EndPoint(host, port)); | ||
| liveBrokers.add(new Broker(brokerId, endPoints, null)); | ||
|
There was a problem hiding this comment. Could use the constructor that doesn't take a rack. |
||
| } else { // V1 or V2 | ||
| Map<SecurityProtocol, EndPoint> endPoints = new HashMap<>(); | ||
| for (Object endPointDataObj : brokerData.getArray(ENDPOINTS_KEY_NAME)) { | ||
| Struct endPointData = (Struct) endPointDataObj; | ||
|
|
@@ -236,11 +246,13 @@ public UpdateMetadataRequest(Struct struct) { | |
| short protocolTypeId = endPointData.getShort(SECURITY_PROTOCOL_TYPE_KEY_NAME); | ||
| endPoints.put(SecurityProtocol.forId(protocolTypeId), new EndPoint(host, port)); | ||
| } | ||
| String rack = null; | ||
| if (brokerData.hasField(RACK_KEY_NAME)) { // V2 | ||
| rack = brokerData.getString(RACK_KEY_NAME); | ||
| } | ||
| liveBrokers.add(new Broker(brokerId, endPoints, rack)); | ||
| } | ||
| } | ||
| controllerId = struct.getInt(CONTROLLER_ID_KEY_NAME); | ||
| controllerEpoch = struct.getInt(CONTROLLER_EPOCH_KEY_NAME); | ||
| this.partitionStates = partitionStates; | ||
|
|
@@ -249,14 +261,11 @@ public UpdateMetadataRequest(Struct struct) { | |
|
|
||
| @Override | ||
| public AbstractRequestResponse getErrorResponse(int versionId, Throwable e) { | ||
| if (versionId <= 2) | ||
| return new UpdateMetadataResponse(Errors.forException(e).code()); | ||
| else | ||
| throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d", | ||
| versionId, this.getClass().getSimpleName(), ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id))); | ||
| } | ||
|
|
||
| public int controllerId() { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should probably mark this as deprecated.