From 58210561106d7f39148a327c52cc2f82863af2e0 Mon Sep 17 00:00:00 2001 From: Akshay Rai Date: Sun, 28 Jun 2026 18:33:36 +0000 Subject: [PATCH] Checkstyle code-safety cleanup: avoid parameter reassignment and null-side equals Bundle of two mechanical code-safety fixes the LinkedIn-internal checkstyle (a stricter superset of the OSS Checker) flags on the current OSS Brooklin tree. No behavior change in any file. Neither rule is currently enabled in OSS Brooklin's own checkstyle.xml, so this PR addresses the latent violations without proposing a config change. 21 fixes across 12 files, by rule: ParameterAssignmentCheck (18) Reassigning a method parameter inside its body is a long-standing readability/debuggability footgun -- the parameter name silently changes meaning partway through the method, debugger watches and stack traces lie, and overloading semantics get confusing for derived classes. The mechanical fix is to introduce a final-by- convention local that captures the transformed value and then use the local for the rest of the method. Touched sites: BaseRestClientFactory.{registerRestClient,addOverride,getOverride} -- 'uri' was being clobbered by sanitizeUri; introduce sanitizedUri local. FieldMetadata.parseMetadata -- 'meta' was being trimmed of its trailing delimiter; introduce trimmedMeta local and fold the if/assign into a ternary. CommonConnectorMetrics.{getEventProcessingMetrics, getEventPollMetrics, getPartitionSpecificMetrics} -- the `prefix = Strings.nullToEmpty(prefix); return X(prefix)` pattern was 3 lines, fold into a single expression. KafkaDestination.parse -- 'uri' was being clobbered by URIUtil.decode inside a try block whose catch still wants to log the original URI; introduce decodedUri local. (No behavior change: the catch already used the pre-decode uri because the assignment is the throwing statement.) AbstractKafkaBasedConnectorTask.containsTransientException -- 'ex' was being walked up the cause chain; rename loop variable to 'cause' so the method's input parameter stays meaningful. KafkaBasedConnectorTaskMetrics.getKafkaBasedConnectorTaskSpecificMetrics -- 'prefix' was being clobbered by Strings.nullToEmpty; introduce safePrefix local used in all 11 metric registrations. DatastreamProducerRecordBuilder.addEvent -- 'key'/'value' were being null-coalesced; introduce eventKey/eventValue locals. ZkClient.ensurePath -- 'path' was being walked up the ZK tree one separator at a time; introduce 'current' local so the parameter is preserved. RestliUtils.sanitizeUri -- 'uri' was being lowercased and potentially scheme-prefixed; introduce 'sanitized' local. RestliUtils.withPaging -- 'stream' was being replaced with .skip()/.limit() variants; introduce 'paged' local. EqualsAvoidNullCheck (3) Calls of the form `variable.equals("literal")` throw NPE when the variable is null, where `"literal".equals(variable)` simply returns false. The fix is to flip the comparison so the string literal is the receiver. Touched sites: DatastreamTaskImpl.getDatastreamTaskName -- `_id.equals("")` -> `"".equals(_id)`. BrokenConnector / DummyConnector ctors -- the dummy-config validation `dummyConfigValue.equals("dummyValue")` -> `"dummyValue".equals(dummyConfigValue)`. (In both cases the variable is currently non-null because it comes from Properties.getProperty(name, ""), but the rule still fires and flipping is harmless.) Motivation outside OSS Brooklin: brooklin-server consumes Brooklin in-tree (subtree-merge at brooklin-core/) and runs a stricter checkstyle on the merged code; the two rules above fire 21 times total today, forcing a build-level carve-out. With this PR merged upstream, brooklin-server's next OSS sync inherits the cleanup and the two rules drop out of its checkstyle-suppress list. --- .../datastream/BaseRestClientFactory.java | 16 ++++++------- .../avrogenerator/FieldMetadata.java | 6 ++--- .../connectors/CommonConnectorMetrics.java | 9 +++---- .../AbstractKafkaBasedConnectorTask.java | 7 +++--- .../kafka/KafkaBasedConnectorTaskMetrics.java | 24 +++++++++---------- .../datastream/kafka/KafkaDestination.java | 5 ++-- .../DatastreamProducerRecordBuilder.java | 6 ++--- .../datastream/server/DatastreamTaskImpl.java | 2 +- .../connectors/BrokenConnector.java | 2 +- .../datastream/connectors/DummyConnector.java | 2 +- .../datastream/common/RestliUtils.java | 15 ++++++------ .../datastream/common/zk/ZkClient.java | 11 +++++---- 12 files changed, 52 insertions(+), 53 deletions(-) diff --git a/datastream-client/src/main/java/com/linkedin/datastream/BaseRestClientFactory.java b/datastream-client/src/main/java/com/linkedin/datastream/BaseRestClientFactory.java index f3b0fb3a3..0c9e186ec 100644 --- a/datastream-client/src/main/java/com/linkedin/datastream/BaseRestClientFactory.java +++ b/datastream-client/src/main/java/com/linkedin/datastream/BaseRestClientFactory.java @@ -113,8 +113,8 @@ public synchronized T getClient(String uri, Map httpConfig, Prop * @param restClient custom R2 RestClient */ public synchronized void registerRestClient(String uri, RestClient restClient) { - uri = RestliUtils.sanitizeUri(uri); - _restClients.put(getKey(uri, Collections.emptyMap()), restClient); + String sanitizedUri = RestliUtils.sanitizeUri(uri); + _restClients.put(getKey(sanitizedUri, Collections.emptyMap()), restClient); } /** @@ -124,16 +124,16 @@ public synchronized void registerRestClient(String uri, RestClient restClient) { */ public synchronized void addOverride(String uri, T restliClient) { Validate.notNull(restliClient, "null RestClient"); - uri = RestliUtils.sanitizeUri(uri); - if (_overrides.containsKey(uri)) { - _logger.warn("Replacing existing RestliClient override for URI: " + uri); + String sanitizedUri = RestliUtils.sanitizeUri(uri); + if (_overrides.containsKey(sanitizedUri)) { + _logger.warn("Replacing existing RestliClient override for URI: " + sanitizedUri); } - _overrides.put(uri, restliClient); + _overrides.put(sanitizedUri, restliClient); } private T getOverride(String uri) { - uri = RestliUtils.sanitizeUri(uri); - return _overrides.getOrDefault(uri, null); + String sanitizedUri = RestliUtils.sanitizeUri(uri); + return _overrides.getOrDefault(sanitizedUri, null); } private static String getKey(String uri, Map httpConfig) { diff --git a/datastream-common/src/main/java/com/linkedin/datastream/avrogenerator/FieldMetadata.java b/datastream-common/src/main/java/com/linkedin/datastream/avrogenerator/FieldMetadata.java index 0936650a5..f9215e0fa 100644 --- a/datastream-common/src/main/java/com/linkedin/datastream/avrogenerator/FieldMetadata.java +++ b/datastream-common/src/main/java/com/linkedin/datastream/avrogenerator/FieldMetadata.java @@ -57,11 +57,9 @@ public FieldMetadata(@NotNull String dbFieldName, String nullable, int dbFieldPo */ public static Map parseMetadata(String meta) { // cut off the trailing delimiter ";" - if (meta.endsWith(META_LIST_DELIMITER)) { - meta = meta.substring(0, meta.length() - 1); - } + String trimmedMeta = meta.endsWith(META_LIST_DELIMITER) ? meta.substring(0, meta.length() - 1) : meta; - String[] parts = meta.split(META_LIST_DELIMITER); + String[] parts = trimmedMeta.split(META_LIST_DELIMITER); Map metas = new HashMap<>(parts.length); for (String part : parts) { String[] keyValuePair = part.split(META_VALUE_DELIMITER); diff --git a/datastream-common/src/main/java/com/linkedin/datastream/connectors/CommonConnectorMetrics.java b/datastream-common/src/main/java/com/linkedin/datastream/connectors/CommonConnectorMetrics.java index e670dfce9..840802e63 100644 --- a/datastream-common/src/main/java/com/linkedin/datastream/connectors/CommonConnectorMetrics.java +++ b/datastream-common/src/main/java/com/linkedin/datastream/connectors/CommonConnectorMetrics.java @@ -492,8 +492,7 @@ public void updateRebalanceRate(long val) { * @return List of exposed Brooklin metrics for the connector activity */ public static List getEventProcessingMetrics(String prefix) { - prefix = Strings.nullToEmpty(prefix); - return EventProcMetrics.getMetricInfos(prefix); + return EventProcMetrics.getMetricInfos(Strings.nullToEmpty(prefix)); } /** @@ -502,8 +501,7 @@ public static List getEventProcessingMetrics(String prefix) * @return List of exposed Brooklin metrics for the connector activity */ public static List getEventPollMetrics(String prefix) { - prefix = Strings.nullToEmpty(prefix); - return PollMetrics.getMetricInfos(prefix); + return PollMetrics.getMetricInfos(Strings.nullToEmpty(prefix)); } /** @@ -512,7 +510,6 @@ public static List getEventPollMetrics(String prefix) { * @return List of exposed Brooklin metrics for the connector activity */ public static List getPartitionSpecificMetrics(String prefix) { - prefix = Strings.nullToEmpty(prefix); - return PartitionMetrics.getMetricInfos(prefix); + return PartitionMetrics.getMetricInfos(Strings.nullToEmpty(prefix)); } } diff --git a/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/AbstractKafkaBasedConnectorTask.java b/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/AbstractKafkaBasedConnectorTask.java index 8e7a92a48..af2ad5856 100644 --- a/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/AbstractKafkaBasedConnectorTask.java +++ b/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/AbstractKafkaBasedConnectorTask.java @@ -351,11 +351,12 @@ protected void rewindAndPausePartitionsOnSendException() { } private boolean containsTransientException(Throwable ex) { - while (ex instanceof DatastreamRuntimeException) { - if (ex instanceof DatastreamTransientException) { + Throwable cause = ex; + while (cause instanceof DatastreamRuntimeException) { + if (cause instanceof DatastreamTransientException) { return true; } - ex = ex.getCause(); + cause = cause.getCause(); } return false; } diff --git a/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaBasedConnectorTaskMetrics.java b/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaBasedConnectorTaskMetrics.java index a1c1c2cf0..7b4cb4ab4 100644 --- a/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaBasedConnectorTaskMetrics.java +++ b/datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaBasedConnectorTaskMetrics.java @@ -324,19 +324,19 @@ public void updatePerEventProcessingTimeNanos(long val) { */ public static List getKafkaBasedConnectorTaskSpecificMetrics(String prefix) { List metrics = new ArrayList<>(); - prefix = Strings.nullToEmpty(prefix); + String safePrefix = Strings.nullToEmpty(prefix); // Specify the attributes to expose to the final metric registry. - metrics.add(new BrooklinGaugeInfo(prefix + NUM_CONFIG_PAUSED_PARTITIONS)); - metrics.add(new BrooklinGaugeInfo(prefix + NUM_AUTO_PAUSED_PARTITIONS_ON_ERROR)); - metrics.add(new BrooklinGaugeInfo(prefix + NUM_AUTO_PAUSED_PARTITIONS_ON_INFLIGHT_MESSAGES)); - metrics.add(new BrooklinGaugeInfo(prefix + NUM_AUTO_PAUSED_PARTITIONS_WAITING_FOR_DEST_TOPIC)); - metrics.add(new BrooklinGaugeInfo(prefix + NUM_AUTO_PAUSED_PARTITIONS_WAITING_SOURCE_TOPIC_ACCESS)); - metrics.add(new BrooklinGaugeInfo(prefix + NUM_TOPICS)); - metrics.add(new BrooklinGaugeInfo(prefix + CONSUMER_OFFSET_WATERMARK_SPAN)); - metrics.add(new BrooklinGaugeInfo(prefix + CONSUMER_LICLOSEST_DATA_LOSS_ESTIMATION)); - metrics.add(new BrooklinHistogramInfo(prefix + POLL_DURATION_MS)); - metrics.add(new BrooklinHistogramInfo(prefix + TIME_SPENT_BETWEEN_POLLS_MS)); - metrics.add(new BrooklinHistogramInfo(prefix + PER_EVENT_PROCESSING_TIME_NANOS)); + metrics.add(new BrooklinGaugeInfo(safePrefix + NUM_CONFIG_PAUSED_PARTITIONS)); + metrics.add(new BrooklinGaugeInfo(safePrefix + NUM_AUTO_PAUSED_PARTITIONS_ON_ERROR)); + metrics.add(new BrooklinGaugeInfo(safePrefix + NUM_AUTO_PAUSED_PARTITIONS_ON_INFLIGHT_MESSAGES)); + metrics.add(new BrooklinGaugeInfo(safePrefix + NUM_AUTO_PAUSED_PARTITIONS_WAITING_FOR_DEST_TOPIC)); + metrics.add(new BrooklinGaugeInfo(safePrefix + NUM_AUTO_PAUSED_PARTITIONS_WAITING_SOURCE_TOPIC_ACCESS)); + metrics.add(new BrooklinGaugeInfo(safePrefix + NUM_TOPICS)); + metrics.add(new BrooklinGaugeInfo(safePrefix + CONSUMER_OFFSET_WATERMARK_SPAN)); + metrics.add(new BrooklinGaugeInfo(safePrefix + CONSUMER_LICLOSEST_DATA_LOSS_ESTIMATION)); + metrics.add(new BrooklinHistogramInfo(safePrefix + POLL_DURATION_MS)); + metrics.add(new BrooklinHistogramInfo(safePrefix + TIME_SPENT_BETWEEN_POLLS_MS)); + metrics.add(new BrooklinHistogramInfo(safePrefix + PER_EVENT_PROCESSING_TIME_NANOS)); return Collections.unmodifiableList(metrics); } } diff --git a/datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaDestination.java b/datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaDestination.java index 184ce979a..b732b37b1 100644 --- a/datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaDestination.java +++ b/datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaDestination.java @@ -50,14 +50,15 @@ public static KafkaDestination parse(String uri) { Validate.isTrue(uri.startsWith(SCHEME_KAFKA) || uri.startsWith(SCHEME_SECURE_KAFKA), "Invalid scheme in URI: " + uri); + String decodedUri; try { // Decode URI in case it's escaped - uri = URIUtil.decode(uri); + decodedUri = URIUtil.decode(uri); } catch (Exception e) { throw new DatastreamRuntimeException("Failed to decode Kafka destination URI: " + uri, e); } - URI u = URI.create(uri); + URI u = URI.create(decodedUri); String scheme = u.getScheme(); String zkAddress = u.getAuthority(); String path = u.getPath(); diff --git a/datastream-server-api/src/main/java/com/linkedin/datastream/server/DatastreamProducerRecordBuilder.java b/datastream-server-api/src/main/java/com/linkedin/datastream/server/DatastreamProducerRecordBuilder.java index cad10c040..b02dd7b50 100644 --- a/datastream-server-api/src/main/java/com/linkedin/datastream/server/DatastreamProducerRecordBuilder.java +++ b/datastream-server-api/src/main/java/com/linkedin/datastream/server/DatastreamProducerRecordBuilder.java @@ -67,9 +67,9 @@ public void setSourceCheckpoint(String sourceCheckpoint) { * Add the event with key and value to the DatastreamProducerRecord. Datastream producer record can have multiple events. */ public void addEvent(Object key, Object value, Object previousValue, Map metadata) { - key = key == null ? new byte[0] : key; - value = value == null ? new byte[0] : value; - _events.add(new BrooklinEnvelope(key, value, previousValue, metadata)); + Object eventKey = key == null ? new byte[0] : key; + Object eventValue = value == null ? new byte[0] : value; + _events.add(new BrooklinEnvelope(eventKey, eventValue, previousValue, metadata)); } /** diff --git a/datastream-server/src/main/java/com/linkedin/datastream/server/DatastreamTaskImpl.java b/datastream-server/src/main/java/com/linkedin/datastream/server/DatastreamTaskImpl.java index 040f76037..7646fd547 100644 --- a/datastream-server/src/main/java/com/linkedin/datastream/server/DatastreamTaskImpl.java +++ b/datastream-server/src/main/java/com/linkedin/datastream/server/DatastreamTaskImpl.java @@ -232,7 +232,7 @@ public String toJson() throws IOException { @JsonIgnore public String getDatastreamTaskName() { - return _id.equals("") ? _taskPrefix : _taskPrefix + "_" + _id; + return "".equals(_id) ? _taskPrefix : _taskPrefix + "_" + _id; } @JsonIgnore diff --git a/datastream-testcommon/src/main/java/com/linkedin/datastream/connectors/BrokenConnector.java b/datastream-testcommon/src/main/java/com/linkedin/datastream/connectors/BrokenConnector.java index e65df8502..61de4b22b 100644 --- a/datastream-testcommon/src/main/java/com/linkedin/datastream/connectors/BrokenConnector.java +++ b/datastream-testcommon/src/main/java/com/linkedin/datastream/connectors/BrokenConnector.java @@ -38,7 +38,7 @@ public class BrokenConnector implements Connector, DiagnosticsAware { public BrokenConnector(Properties properties) throws Exception { _properties = properties; String dummyConfigValue = _properties.getProperty("dummyProperty", ""); - if (!dummyConfigValue.equals("dummyValue")) { + if (!"dummyValue".equals(dummyConfigValue)) { throw new Exception("Invalid config value for dummyProperty. Expected: dummyValue"); } } diff --git a/datastream-testcommon/src/main/java/com/linkedin/datastream/connectors/DummyConnector.java b/datastream-testcommon/src/main/java/com/linkedin/datastream/connectors/DummyConnector.java index 6a7ae8fad..a18b203c8 100644 --- a/datastream-testcommon/src/main/java/com/linkedin/datastream/connectors/DummyConnector.java +++ b/datastream-testcommon/src/main/java/com/linkedin/datastream/connectors/DummyConnector.java @@ -42,7 +42,7 @@ public class DummyConnector implements Connector, DiagnosticsAware { public DummyConnector(Properties properties) throws Exception { _properties = properties; String dummyConfigValue = _properties.getProperty("dummyProperty", ""); - if (!dummyConfigValue.equals("dummyValue")) { + if (!"dummyValue".equals(dummyConfigValue)) { throw new Exception("Invalid config value for dummyProperty. Expected: dummyValue"); } } diff --git a/datastream-utils/src/main/java/com/linkedin/datastream/common/RestliUtils.java b/datastream-utils/src/main/java/com/linkedin/datastream/common/RestliUtils.java index 5f5c57b9d..4502d74a4 100644 --- a/datastream-utils/src/main/java/com/linkedin/datastream/common/RestliUtils.java +++ b/datastream-utils/src/main/java/com/linkedin/datastream/common/RestliUtils.java @@ -26,12 +26,12 @@ public final class RestliUtils { * a {@code "/"} appended to it unless it already has one. */ public static String sanitizeUri(String uri) { - uri = uri.toLowerCase(); - if (!uri.startsWith(URI_SCHEME) && !uri.startsWith(URI_SCHEME_SSL)) { + String sanitized = uri.toLowerCase(); + if (!sanitized.startsWith(URI_SCHEME) && !sanitized.startsWith(URI_SCHEME_SSL)) { // use plaintext by default - uri = URI_SCHEME + uri; + sanitized = URI_SCHEME + sanitized; } - return StringUtils.appendIfMissing(uri, "/"); + return StringUtils.appendIfMissing(sanitized, "/"); } /** @@ -41,14 +41,15 @@ public static String sanitizeUri(String uri) { * @param type of elements in the Stream */ public static Stream withPaging(Stream stream, final PagingContext pagingContext) { + Stream paged = stream; if (pagingContext.hasStart()) { - stream = stream.skip(pagingContext.getStart()); + paged = paged.skip(pagingContext.getStart()); } if (pagingContext.hasCount()) { - stream = stream.limit(pagingContext.getCount()); + paged = paged.limit(pagingContext.getCount()); } - return stream; + return paged; } } diff --git a/datastream-utils/src/main/java/com/linkedin/datastream/common/zk/ZkClient.java b/datastream-utils/src/main/java/com/linkedin/datastream/common/zk/ZkClient.java index 96e584a46..fe4d44544 100644 --- a/datastream-utils/src/main/java/com/linkedin/datastream/common/zk/ZkClient.java +++ b/datastream-utils/src/main/java/com/linkedin/datastream/common/zk/ZkClient.java @@ -173,11 +173,12 @@ public void ensurePath(String path) { Stack pstack = new Stack<>(); // push paths in stack because we need to create from parent to children - while (!exists(path)) { - pstack.push(path); - path = path.substring(0, path.lastIndexOf(ZK_PATH_SEPARATOR)); - if (path.isEmpty()) { - path = ZK_PATH_SEPARATOR; + String current = path; + while (!exists(current)) { + pstack.push(current); + current = current.substring(0, current.lastIndexOf(ZK_PATH_SEPARATOR)); + if (current.isEmpty()) { + current = ZK_PATH_SEPARATOR; } }