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; } }