github-actions[bot] commented on code in PR #69391:
URL: https://github.com/apache/doris/pull/69391#discussion_r4358294944
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -790,7 +809,9 @@ public void modifyProperties(AlterRoutineLoadCommand
command) throws UserExcepti
throw new DdlException("Only supports modification of PAUSED
jobs");
}
- modifyPropertiesInternal(jobProperties, dataSourceProperties);
+ Map<String, String> convertedProperties =
prepareModifyProperties(dataSourceProperties);
+ applyProperties(jobProperties, dataSourceProperties);
+ setConvertedCustomProperties(convertedProperties);
Review Comment:
[P1] Preserve datetime defaults when reusing the conversion cache
For a job created with a datetime `property.kafka_default_offsets`, the
first conversion removes `kafka_origin_default_offsets` but leaves the numeric
`kafka_default_offsets` in the cache. A job-only ALTER (for example, changing
`max_batch_rows`) returns that existing cache from `prepareModifyProperties()`,
so this unconditional setter extracts the remaining millis string into
`kafkaDefaultOffSet`. `convertedDefaultOffsetToLong()` then falls back to
`OFFSET_END`. If a partition is added while the job is paused and the job
resumes, the scheduler updates partition progress before `prepare()` rebuilds
the cache, records that partition's end offset, and skips its records since the
configured datetime. Extract defaults only from a freshly resolved map, or
preserve the derived default when reusing the cache.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -805,21 +826,37 @@ private void convertOffset(KafkaDataSourceProperties
dataSourceProperties) throw
if (partitionOffsets.isEmpty()) {
return;
}
+ Map<String, String> properties;
+ String brokerListSnapshot;
+ String topicSnapshot;
+ writeLock();
+ try {
+ // Recovery leaves the conversion cache empty. Validate the
effective
+ // properties before an ALTER offset lookup can contact an older
BE.
+ convertCustomProperties(false);
Review Comment:
[P2] Merge repairing ALTER properties before validating offset lookups
An older paused job with `ssl.ca.location=/etc/ca.pem` can be migrated to
`FILE:ca.pem`, but adding `kafka_partitions=0` and `kafka_offsets=100` to that
same valid ALTER makes this call reject the old raw-path map.
`modifyProperties()` invokes `convertOffset()` before
`prepareModifyProperties()` merges the replacement, so the repair never reaches
validation. This also occurs for explicit positive offsets, where
`KafkaUtil.getRealOffsets()` needs no BE request at all. Prepare and validate
the merged effective properties first, then use that safe snapshot for any
required offset lookup while preserving atomic publication.
##########
fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaPropertyValidationTest.java:
##########
@@ -0,0 +1,413 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load.routineload;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.cloud.rpc.MetaServiceProxy;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.Pair;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.util.KafkaUtil;
Review Comment:
[P1] Import the existing Kafka utility and ALTER command classes
This new test imports `org.apache.doris.common.util.KafkaUtil` and
`org.apache.doris.nereids.trees.plans.commands.load.AlterRoutineLoadCommand`,
but neither class exists in those packages on this head. The actual classes are
`org.apache.doris.datasource.kafka.KafkaUtil` and
`org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand`, as
used by `KafkaRoutineLoadJob`. Both types are referenced by these tests, so FE
test compilation fails before any of the new validation/recovery tests can run.
Correct both imports.
##########
fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaPropertyValidationTest.java:
##########
@@ -0,0 +1,413 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load.routineload;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.cloud.rpc.MetaServiceProxy;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.Pair;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.util.KafkaUtil;
+import org.apache.doris.common.util.SmallFileMgr;
+import org.apache.doris.common.util.SmallFileMgr.SmallFile;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.load.routineload.kafka.KafkaDataSourceProperties;
+import org.apache.doris.load.routineload.kafka.KafkaProgress;
+import org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob;
+import org.apache.doris.nereids.parser.NereidsParser;
+import
org.apache.doris.nereids.trees.plans.commands.load.AlterRoutineLoadCommand;
+import
org.apache.doris.nereids.trees.plans.commands.load.CreateRoutineLoadCommand;
+import org.apache.doris.persist.AlterRoutineLoadJobOperationLog;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.RoutineLoadOperation;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.thrift.BackendServiceConstants;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public class KafkaPropertyValidationTest {
+ private KafkaDataSourceProperties properties(String key, String value,
boolean alter) {
+ Map<String, String> source = new HashMap<>();
+ if (!alter) {
+ source.put("kafka_broker_list", "127.0.0.1:9092");
+ source.put("kafka_topic", "test_topic");
+ }
+ source.put("property." + key, value);
+ KafkaDataSourceProperties properties = new
KafkaDataSourceProperties(source);
+ properties.setAlter(alter);
+ properties.setTimezone("UTC");
+ return properties;
+ }
+
+ @Test
+ public void testCreateAndAlterRejectExecutableAndUnknownProperties() {
+ for (boolean alter : new boolean[] {false, true}) {
+ for (String key : List.of("sasl.kerberos.kinit.cmd",
"sasl.kerberos.keytab",
+ "sasl.kerberos.min.time.before.relogin",
"plugin.library.paths", "ssl.engine.location",
+ "ssl.engine.id", "ssl.providers",
"unknown.future.property", "sasl.jaas.config",
+ "security.protocol\0suffix")) {
+ AnalysisException error =
Assertions.assertThrows(AnalysisException.class,
+ () -> properties(key, "sensitive-value",
alter).analyze());
+ Assertions.assertTrue(error.getMessage().contains("is not
allowed in Routine Load"), key);
+
Assertions.assertFalse(error.getMessage().contains("sensitive-value"));
+ }
+ }
+ }
+
+ @Test
+ public void testRejectKerberosRefreshReenable() {
+ for (boolean alter : new boolean[] {false, true}) {
+ for (String value : List.of("1", "60000", "00", "0\0suffix",
"FILE:interval")) {
+ Assertions.assertThrows(AnalysisException.class,
+ () ->
properties("sasl.kerberos.min.time.before.relogin", value, alter).analyze());
+ }
+ }
+ }
+
+ @Test
+ public void testCertificateFilesRequireManagedReferences() throws
Exception {
+ for (boolean alter : new boolean[] {false, true}) {
+ for (String key : BackendServiceConstants.KAFKA_FILE_PROPERTIES) {
+ for (String value : List.of("/etc/ssl/private/key.pem",
"relative.pem", "", "FILE:")) {
+ AnalysisException error =
Assertions.assertThrows(AnalysisException.class,
+ () -> properties(key, value, alter).analyze());
+ Assertions.assertTrue(error.getMessage().contains("CREATE
FILE"));
+ }
+ KafkaDataSourceProperties properties = properties(key,
"FILE:client.pem", alter);
+ properties.analyze();
+ Assertions.assertEquals("FILE:client.pem",
properties.getCustomKafkaProperties().get(key));
+ }
+ }
+ }
+
+ @Test
+ public void testNormalConsumerPropertiesAndDefaultOffsets() throws
Exception {
+ Map<String, String> accepted = Map.of(
+ "group.id", "routine-load-group",
+ "client.id", "routine-load-client",
+ "security.protocol", "SASL_SSL",
+ "sasl.mechanisms", "SCRAM-SHA-256",
+ "sasl.username", "user",
+ "sasl.password", "secret",
+ "isolation.level", "read_committed",
+ "fetch.max.bytes", "1048576",
+ "kafka_default_offsets", "OFFSET_BEGINNING");
+ for (boolean alter : new boolean[] {false, true}) {
+ for (Map.Entry<String, String> property : accepted.entrySet()) {
+ properties(property.getKey(), property.getValue(),
alter).analyze();
+ }
+ }
+ }
+
+ @Test
+ public void testAwsPropertiesWithAndWithoutPrefix() throws Exception {
+ for (String prefix : List.of("", "property.")) {
+ Map<String, String> source = new HashMap<>();
+ source.put("kafka_broker_list", "127.0.0.1:9098");
+ source.put("kafka_topic", "test_topic");
+ source.put("property.security.protocol", "SASL_SSL");
+ source.put("property.sasl.mechanism", "OAUTHBEARER");
+ source.put(prefix + "aws.region", "us-east-1");
+ source.put(prefix + "aws.role_arn",
"arn:aws:iam::123456789012:role/test");
+ KafkaDataSourceProperties properties = new
KafkaDataSourceProperties(source);
+ properties.setTimezone("UTC");
+ properties.analyze();
+ Assertions.assertEquals("us-east-1",
properties.getCustomKafkaProperties().get("aws.region"));
+ }
+ }
+
+ @Test
+ public void testDisableKerberosCommandForOlderBackends() throws Exception {
+ properties("sasl.kerberos.min.time.before.relogin", "0",
false).analyze();
+ for (Map<String, String> custom : List.of(Map.<String, String>of(),
+ Map.of("group.id", "test", "sasl.kerberos.principal",
"user;unused"),
+ Map.of("kafka_default_offsets", "OFFSET_BEGINNING",
+ "kafka_origin_default_offsets", "OFFSET_BEGINNING"))) {
+ KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job",
1L, 1L,
+ "127.0.0.1:9092", "test_topic", UserIdentity.ADMIN);
+ Deencapsulation.setField(job, "customProperties", custom);
+ Deencapsulation.invoke(job, "convertCustomProperties", true);
+ Assertions.assertEquals("0", job.getConvertedCustomProperties()
+ .get("sasl.kerberos.min.time.before.relogin"));
+ }
+ }
+
+ @Test
+ public void testPersistedPropertiesRejectedBeforeMetadataRequests() {
+ KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job", 1L,
1L,
+ "127.0.0.1:9092", "test_topic", UserIdentity.ADMIN);
+ Deencapsulation.setField(job, "customProperties",
Map.of("plugin.library.paths", "unused.so"));
+ // Even a previously populated conversion cache must not bypass
validation.
+ Deencapsulation.setField(job, "convertedCustomProperties",
Map.of("group.id", "cached"));
+ DdlException error = Assertions.assertThrows(DdlException.class,
+ () -> Deencapsulation.invoke(job, "convertCustomProperties",
false));
+ Assertions.assertTrue(error.getMessage().contains("is not allowed in
Routine Load"));
+ }
+
+ // Build an already analyzed, persisted ALTER from an older FE without
applying
+ // today's CREATE/ALTER admission rules to the historical input.
+ private AlterRoutineLoadJobOperationLog historicalAlter(Map<String,
String> custom,
+ String topic, String brokers, List<Pair<Integer, Long>> offsets) {
+ Map<String, String> source = new HashMap<>();
+ custom.forEach((key, value) -> source.put("property." + key, value));
+ source.put("kafka_topic", topic);
+ source.put("kafka_broker_list", brokers);
+ KafkaDataSourceProperties properties = new
KafkaDataSourceProperties(source);
+ properties.setAlter(true);
+ Deencapsulation.setField(properties, "dataSourceType", "KAFKA");
+ Deencapsulation.setField(properties, "customKafkaProperties", custom);
+ Deencapsulation.setField(properties, "topic", topic);
+ Deencapsulation.setField(properties, "brokerList", brokers);
+ properties.setKafkaPartitionOffsets(offsets);
+ AlterRoutineLoadJobOperationLog log = new
AlterRoutineLoadJobOperationLog(1L,
+ Map.of("max_batch_rows", "2000000"), properties);
+ return GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(log),
AlterRoutineLoadJobOperationLog.class);
+ }
+
+ @Test
+ public void testReplayHistoricalAlterRestoresCompleteState() throws
Exception {
+ for (boolean cloud : new boolean[] {false, true}) {
+ try (MockedStatic<Config> config =
Mockito.mockStatic(Config.class);
+ MockedStatic<Env> env = Mockito.mockStatic(Env.class);
+ MockedStatic<MetaServiceProxy> metaService =
Mockito.mockStatic(MetaServiceProxy.class);
+ MockedStatic<KafkaUtil> kafka =
Mockito.mockStatic(KafkaUtil.class)) {
+ config.when(Config::isCloudMode).thenReturn(cloud);
+ KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L,
"test_job", 1L, 2L,
+ "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN);
+ Deencapsulation.setField(job, "progress", new
KafkaProgress(Map.of(0, 5L, 1, 6L)));
+ job.prepare();
+
+ AlterRoutineLoadJobOperationLog log = historicalAlter(
+ Map.of("sasl.kerberos.min.time.before.relogin",
"60000"),
+ "topic_b", "127.0.0.2:9092", List.of(Pair.of(0,
100L)));
+ job.replayModifyProperties(log);
+
+ Assertions.assertEquals("topic_b", job.getTopic());
+ Assertions.assertEquals("127.0.0.2:9092", job.getBrokerList());
+ Assertions.assertEquals(Map.of(0, 100L),
+ ((KafkaProgress)
job.getProgress()).getOffsetByPartition());
+ Assertions.assertEquals(2000000L, job.getMaxBatchRows());
+ Assertions.assertEquals("60000", job.getCustomProperties()
+
.get("property.sasl.kerberos.min.time.before.relogin"));
+
Assertions.assertTrue(job.getConvertedCustomProperties().isEmpty());
+
+ DdlException error =
Assertions.assertThrows(DdlException.class, job::prepare);
+ Assertions.assertTrue(error.getMessage().contains("Kerberos
tickets must be managed externally"));
+ Assertions.assertThrows(DdlException.class,
+ () -> Deencapsulation.invoke(job,
"getAllKafkaPartitions"));
+ Assertions.assertThrows(DdlException.class,
+ () -> Deencapsulation.invoke(job, "convertOffset",
log.getDataSourceProperties()));
+
+ // A later committed repair must retain topic B and its
offsets.
+ job.replayModifyProperties(historicalAlter(
+ Map.of("sasl.kerberos.min.time.before.relogin", "0"),
"", "", List.of()));
+ job.prepare();
+ Assertions.assertEquals("topic_b", job.getTopic());
+ Assertions.assertEquals("127.0.0.2:9092", job.getBrokerList());
+ Assertions.assertEquals(Map.of(0, 100L),
+ ((KafkaProgress)
job.getProgress()).getOffsetByPartition());
+ Assertions.assertEquals("0", job.getConvertedCustomProperties()
+ .get("sasl.kerberos.min.time.before.relogin"));
+ env.verifyNoInteractions();
+ metaService.verifyNoInteractions();
+ kafka.verifyNoInteractions();
+ }
+ }
+ }
+
+ @Test
+ public void testReplayDefersManagedFileResolution() throws Exception {
+ Env env = Mockito.mock(Env.class);
+ SmallFileMgr smallFileMgr = Mockito.mock(SmallFileMgr.class);
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getSmallFileMgr()).thenReturn(smallFileMgr);
+ Mockito.when(smallFileMgr.getSmallFile(1L, "kafka", "missing.pem",
true))
+ .thenThrow(new DdlException("missing managed file"));
+ KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job",
1L, 2L,
+ "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN);
+
+ // Partition 3 need not exist in the pre-replay progress. These
offsets
+ // were checked when the ALTER was committed, not during recovery.
+
job.replayModifyProperties(historicalAlter(Map.of("ssl.ca.location",
"FILE:missing.pem"),
+ "topic_b", "127.0.0.2:9092", List.of(Pair.of(3, 100L))));
+ Assertions.assertEquals("topic_b", job.getTopic());
+ Assertions.assertEquals(Map.of(3, 100L),
+ ((KafkaProgress)
job.getProgress()).getOffsetByPartition());
+ Assertions.assertEquals(2000000L, job.getMaxBatchRows());
+ envStatic.verifyNoInteractions();
+ Mockito.verifyNoInteractions(smallFileMgr);
+
+ DdlException error = Assertions.assertThrows(DdlException.class,
job::prepare);
+ Assertions.assertTrue(error.getMessage().contains("missing managed
file"));
+
Assertions.assertTrue(job.getConvertedCustomProperties().isEmpty());
+ }
+ }
+
+ @Test
+ public void testSchedulerPausesReplayedJobWithRejectedProperties() throws
Exception {
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ RoutineLoadManager manager = Mockito.mock(RoutineLoadManager.class);
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class);
+ MockedStatic<KafkaUtil> kafka =
Mockito.mockStatic(KafkaUtil.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job",
1L, 2L,
+ "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN);
+
job.replayModifyProperties(historicalAlter(Map.of("plugin.library.paths",
"unused.so"),
+ "topic_b", "127.0.0.2:9092", List.of()));
+
Mockito.when(manager.getRoutineLoadJobByState(Mockito.anySet())).thenReturn(List.of(job));
+
+ Deencapsulation.invoke(new RoutineLoadScheduler(manager),
"process");
+
+ Assertions.assertEquals(RoutineLoadJob.JobState.PAUSED,
job.getState());
+
Assertions.assertTrue(job.getPauseReason().getMsg().contains("plugin.library.paths"));
+ Assertions.assertEquals("topic_b", job.getTopic());
+ kafka.verifyNoInteractions();
+
Mockito.verify(editLog).logOpRoutineLoadJob(Mockito.any(RoutineLoadOperation.class));
+ }
+ }
+
+ @Test
+ public void testLiveAlterPublishesOnlyPreparedProperties() throws
Exception {
+ Env env = Mockito.mock(Env.class);
+ SmallFileMgr smallFileMgr = Mockito.mock(SmallFileMgr.class);
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getSmallFileMgr()).thenReturn(smallFileMgr);
+ Mockito.when(smallFileMgr.getSmallFile(1L, "kafka", "missing.pem",
true))
+ .thenThrow(new DdlException("missing managed file"));
+ KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(1L, "test_job",
1L, 2L,
+ "127.0.0.1:9092", "topic_a", UserIdentity.ADMIN);
+ Deencapsulation.setField(job, "state",
RoutineLoadJob.JobState.PAUSED);
+ job.prepare();
+ Map<String, String> converted = new
HashMap<>(job.getConvertedCustomProperties());
+ AlterRoutineLoadJobOperationLog log =
historicalAlter(Map.of("ssl.ca.location", "FILE:missing.pem"),
+ "topic_b", "127.0.0.2:9092", List.of());
+ AlterRoutineLoadCommand command =
Mockito.mock(AlterRoutineLoadCommand.class);
+
Mockito.when(command.getAnalyzedJobProperties()).thenReturn(log.getJobProperties());
+
Mockito.when(command.getDataSourceProperties()).thenReturn(log.getDataSourceProperties());
+
+ Assertions.assertThrows(DdlException.class, () ->
job.modifyProperties(command));
+ Assertions.assertEquals("topic_a", job.getTopic());
+ Assertions.assertEquals("127.0.0.1:9092", job.getBrokerList());
+ Assertions.assertTrue(job.getCustomProperties().isEmpty());
+ Assertions.assertEquals(converted,
job.getConvertedCustomProperties());
+ Mockito.verify(env, Mockito.never()).getEditLog();
+
+ KafkaDataSourceProperties safe =
properties("sasl.kerberos.min.time.before.relogin", "0", true);
+ safe.analyze();
+ Mockito.when(command.getDataSourceProperties()).thenReturn(safe);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ job.modifyProperties(command);
+
Assertions.assertEquals(Map.of("property.sasl.kerberos.min.time.before.relogin",
"0"),
Review Comment:
[P2] Include the analyzed default offset in the ALTER expectation
After correcting the imports, this assertion still fails: `safe.analyze()`
calls `analyzeKafkaDefaultOffsetProperty()`, which inserts
`kafka_default_offsets=OFFSET_END` even for this ALTER. `applyProperties()`
copies that entry into the raw custom map, and `getCustomProperties()` returns
it as `property.kafka_default_offsets`; only the converted map removes the
bookkeeping key. The actual map therefore has two entries, while this assertion
expects one. Include the inserted default-offset property, or assert the
intended entry without comparing against an incomplete map.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
