lucasbru commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4412318571
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology
fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting
or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic
indexes are compared
+ * without regard to order, but duplicate occurrences must still match.
All other lists
+ * retain their order-sensitive comparison, including the topic lists
referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology)
{
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested :
topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject
duplicates so a
+ // malformed request cannot match one stored subtopology twice and
omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology
subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its
lists
+ // may be unmodifiable or shared with the heartbeat request.
+ Subtopology copy = subtopology.duplicate();
+
+ for (CopartitionGroup group : copy.copartitionGroups()) {
+ Collections.sort(group.sourceTopics());
+ Collections.sort(group.sourceTopicRegex());
+ Collections.sort(group.repartitionSourceTopics());
+ }
+
+ copy.copartitionGroups().sort(
+ Comparator
+ .comparing(
+ CopartitionGroup::sourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ CopartitionGroup::sourceTopicRegex,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ CopartitionGroup::repartitionSourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ );
+ return copy;
+ }
+
+ private static int compareTopicIndexes(List<Short> first, List<Short>
second) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by
StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current,
StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ &&
current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(),
requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(),
requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(),
requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current,
List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested =
requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() !=
fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(),
fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly
constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(),
requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol
version
Review Comment:
I'd drop this TODO, or move it to a JIRA.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java:
##########
@@ -122,6 +132,233 @@ public void
fromHeartbeatRequestShouldCreateCorrectTopology() {
assertEquals(mkSubtopology2(),
topology.subtopologies().get(SUBTOPOLOGY_ID_2));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void
fromRecordShouldNormalizeCopartitionGroupsWithoutMutation(String topicType) {
+ StreamsGroupTopologyValue alreadySortedExpected =
StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(
+ topologyWithCopartitionGroups(topicType)
+ );
+
+ StreamsGroupTopologyValue record = alreadySortedExpected.duplicate();
+ List<StreamsGroupTopologyValue.CopartitionGroup> groups =
record.subtopologies().get(0).copartitionGroups();
+ Collections.reverse(groups);
+ for (StreamsGroupTopologyValue.CopartitionGroup group : groups) {
+ Collections.reverse(group.sourceTopics());
+ Collections.reverse(group.sourceTopicRegex());
+ Collections.reverse(group.repartitionSourceTopics());
+ }
+ StreamsGroupTopologyValue originalRecord = record.duplicate();
+
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+
+ Subtopology expectedSubtopology =
alreadySortedExpected.subtopologies().get(0);
+ assertEquals(
+ expectedSubtopology,
+ replayed.subtopologies().get(expectedSubtopology.subtopologyId())
+ );
+ assertEquals(originalRecord, record);
+ }
+
+
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void reorderedRequestShouldMatchWithoutMutation(String topicType) {
+ StreamsGroupHeartbeatRequestData.Topology request =
+ topologyWithCopartitionGroups(topicType);
+ StreamsTopology stored = StreamsTopology.fromRecord(
+
StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request)
+ );
+
+ // Stored groups: [[0], [0, 1], [2, 3]]
+ // Request groups: [[3, 2], [1, 0], [0]]
+ StreamsGroupHeartbeatRequestData.Topology reorderedRequest =
request.duplicate();
+ reorderedRequest.subtopologies().get(0).setCopartitionGroups(List.of(
+ copartitionGroup(topicType, List.of((short) 3, (short) 2)),
+ copartitionGroup(topicType, List.of((short) 1, (short) 0)),
+ copartitionGroup(topicType, List.of((short) 0))
+ ));
+ StreamsGroupHeartbeatRequestData.Topology originalRequest =
+ reorderedRequest.duplicate();
+
+ assertTrue(stored.matches(reorderedRequest));
+ assertEquals(originalRequest, reorderedRequest);
+ }
+
+
+ @Test
+ public void mixedCopartitionGroupsShouldMatchRegardlessOfOrder() {
+ // request has 'source - [1] / repartition - [3, 2]' and 'source -
[1], repartition - [0]'
+ // reordered has 'source - [1], repartition - [0' and 'source - [1],
repartition = [2, 3]'
+ StreamsGroupHeartbeatRequestData.Topology request =
topologyWithCopartitionGroups("source");
+ request.subtopologies().get(0).setCopartitionGroups(List.of(
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setSourceTopics(List.of((short) 1))
+ .setRepartitionSourceTopics(List.of((short) 3, (short) 2)),
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setSourceTopics(List.of((short) 1))
+ .setRepartitionSourceTopics(List.of((short) 0))
+ ));
+
+ StreamsGroupTopologyValue record =
StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request);
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+
+ StreamsGroupHeartbeatRequestData.Topology reorderedRequest =
request.duplicate();
+
Collections.reverse(reorderedRequest.subtopologies().get(0).copartitionGroups());
+ reorderedRequest.subtopologies().get(0).copartitionGroups().get(1)
+ .setRepartitionSourceTopics(List.of((short) 2, (short) 3));
+
+ assertTrue(replayed.matches(reorderedRequest));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void differentCopartitionConstraintsShouldNotMatch(String
topicType) {
+ // request - [[0], [0,1], [2,3]]
+ // changed - [[0], [0,1], [1,3]]
+ StreamsGroupHeartbeatRequestData.Topology request =
topologyWithCopartitionGroups(topicType);
+ StreamsTopology original =
StreamsTopology.fromHeartbeatRequest(request);
+
+ StreamsGroupHeartbeatRequestData.Topology changed =
request.duplicate();
+ changed.subtopologies()
+ .get(0)
+ .copartitionGroups()
+ .set(2, copartitionGroup(topicType, List.of((short) 1, (short)
3)));
+
+ assertFalse(original.matches(changed));
+ }
+
+ private static Stream<Arguments> nonCopartitionFieldChanges() {
Review Comment:
nit: this could be a plain `Stream.of(Arguments.of("source topic",
(Consumer<...>) t -> ...), ...)` without the intermediate map.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology
fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting
or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic
indexes are compared
+ * without regard to order, but duplicate occurrences must still match.
All other lists
+ * retain their order-sensitive comparison, including the topic lists
referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology)
{
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested :
topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject
duplicates so a
+ // malformed request cannot match one stored subtopology twice and
omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology
subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its
lists
+ // may be unmodifiable or shared with the heartbeat request.
+ Subtopology copy = subtopology.duplicate();
+
+ for (CopartitionGroup group : copy.copartitionGroups()) {
+ Collections.sort(group.sourceTopics());
+ Collections.sort(group.sourceTopicRegex());
+ Collections.sort(group.repartitionSourceTopics());
+ }
+
+ copy.copartitionGroups().sort(
+ Comparator
+ .comparing(
+ CopartitionGroup::sourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ CopartitionGroup::sourceTopicRegex,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ CopartitionGroup::repartitionSourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ );
+ return copy;
+ }
+
+ private static int compareTopicIndexes(List<Short> first, List<Short>
second) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by
StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current,
StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ &&
current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(),
requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(),
requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(),
requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current,
List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested =
requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() !=
fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(),
fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly
constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(),
requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol
version
+ // requires canonical copartition ordering and older versions are no
longer
+ // supported. Validate the required ordering before comparing.
+
+ // Copy before sorting to preserve the request and its index lists.
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup>
sortedRequested =
+ new ArrayList<>(requested.size());
+
+ for (StreamsGroupHeartbeatRequestData.CopartitionGroup group :
requested) {
+ StreamsGroupHeartbeatRequestData.CopartitionGroup copy =
group.duplicate();
+ Collections.sort(copy.sourceTopics());
+ Collections.sort(copy.sourceTopicRegex());
+ Collections.sort(copy.repartitionSourceTopics());
+ sortedRequested.add(copy);
+ }
+
+ sortedRequested.sort(
Review Comment:
The sorting is now duplicated for the two `CopartitionGroup` types. What
about converting the request's groups to
`StreamsGroupTopologyValue.CopartitionGroup` copies and reusing the sort from
`normalizeCopartitionGroups`? Then we can compare the lists with `equals` and
drop `matchesCopartitionGroup`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology
fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting
or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic
indexes are compared
+ * without regard to order, but duplicate occurrences must still match.
All other lists
+ * retain their order-sensitive comparison, including the topic lists
referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology)
{
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested :
topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject
duplicates so a
+ // malformed request cannot match one stored subtopology twice and
omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology
subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its
lists
+ // may be unmodifiable or shared with the heartbeat request.
+ Subtopology copy = subtopology.duplicate();
+
+ for (CopartitionGroup group : copy.copartitionGroups()) {
+ Collections.sort(group.sourceTopics());
+ Collections.sort(group.sourceTopicRegex());
+ Collections.sort(group.repartitionSourceTopics());
+ }
+
+ copy.copartitionGroups().sort(
+ Comparator
+ .comparing(
+ CopartitionGroup::sourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ CopartitionGroup::sourceTopicRegex,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ CopartitionGroup::repartitionSourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ );
+ return copy;
+ }
+
+ private static int compareTopicIndexes(List<Short> first, List<Short>
second) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by
StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current,
StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ &&
current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(),
requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(),
requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(),
requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current,
List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested =
requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() !=
fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(),
fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly
constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(),
requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol
version
+ // requires canonical copartition ordering and older versions are no
longer
+ // supported. Validate the required ordering before comparing.
+
+ // Copy before sorting to preserve the request and its index lists.
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup>
sortedRequested =
+ new ArrayList<>(requested.size());
+
+ for (StreamsGroupHeartbeatRequestData.CopartitionGroup group :
requested) {
+ StreamsGroupHeartbeatRequestData.CopartitionGroup copy =
group.duplicate();
+ Collections.sort(copy.sourceTopics());
+ Collections.sort(copy.sourceTopicRegex());
+ Collections.sort(copy.repartitionSourceTopics());
+ sortedRequested.add(copy);
+ }
+
+ sortedRequested.sort(
+ Comparator.comparing(
+
StreamsGroupHeartbeatRequestData.CopartitionGroup::sourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+
StreamsGroupHeartbeatRequestData.CopartitionGroup::sourceTopicRegex,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+
StreamsGroupHeartbeatRequestData.CopartitionGroup::repartitionSourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ );
+
+ // Stored groups were already sorted by fromRecord.
Review Comment:
This assumes the stored groups are sorted, but `fromHeartbeatRequest` builds
a topology without going through `fromRecord`. Could it just delegate to
`fromRecord(convertToStreamsGroupTopologyRecord(topology))`? Then the test
setups using it get the same normalization, and removing it later becomes
optional.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
