From c93652b46d0b2d47808ea2b4d392606a1e5d4fd8 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 7 Jun 2024 14:33:13 +0800 Subject: [PATCH] Style improvement --- .../persistence/pipe/PipePluginInfo.java | 39 +++++++++------- .../executor/ClusterConfigTaskExecutor.java | 45 ++++++++++--------- .../resources/conf/iotdb-system.properties | 2 +- .../iotdb/commons/conf/CommonDescriptor.java | 5 +-- .../connector/protocol/IoTDBConnector.java | 28 +++++++----- 5 files changed, 66 insertions(+), 53 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java index 382ebd51620e8..d3c46c2957d7b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java @@ -91,7 +91,8 @@ public void releasePipePluginInfoLock() { /////////////////////////////// Validator /////////////////////////////// - public void validateBeforeCreatingPipePlugin(String pluginName, String jarName, String jarMD5) { + public void validateBeforeCreatingPipePlugin( + final String pluginName, final String jarName, final String jarMD5) { // both build-in and user defined pipe plugin should be unique if (pipePluginMetaKeeper.containsPipePlugin(pluginName)) { throw new PipeException( @@ -108,9 +109,13 @@ public void validateBeforeCreatingPipePlugin(String pluginName, String jarName, } } - public void validateBeforeDroppingPipePlugin(String pluginName) { - if (pipePluginMetaKeeper.containsPipePlugin(pluginName) - && pipePluginMetaKeeper.getPipePluginMeta(pluginName).isBuiltin()) { + public void validateBeforeDroppingPipePlugin(final String pluginName) { + if (!pipePluginMetaKeeper.containsPipePlugin(pluginName)) { + throw new PipeException( + String.format( + "Failed to drop PipePlugin [%s], this PipePlugin has not been created", pluginName)); + } + if (pipePluginMetaKeeper.getPipePluginMeta(pluginName).isBuiltin()) { throw new PipeException( String.format( "Failed to drop PipePlugin [%s], the PipePlugin is a built-in PipePlugin", @@ -118,14 +123,14 @@ public void validateBeforeDroppingPipePlugin(String pluginName) { } } - public boolean isJarNeededToBeSavedWhenCreatingPipePlugin(String jarName) { + public boolean isJarNeededToBeSavedWhenCreatingPipePlugin(final String jarName) { return !pipePluginMetaKeeper.containsJar(jarName); } public void checkPipePluginExistence( - Map extractorAttributes, - Map processorAttributes, - Map connectorAttributes) { + final Map extractorAttributes, + final Map processorAttributes, + final Map connectorAttributes) { final PipeParameters extractorParameters = new PipeParameters(extractorAttributes); final String extractorPluginName = extractorParameters.getStringOrDefault( @@ -170,7 +175,7 @@ public void checkPipePluginExistence( /////////////////////////////// Pipe Plugin Management /////////////////////////////// - public TSStatus createPipePlugin(CreatePipePluginPlan createPipePluginPlan) { + public TSStatus createPipePlugin(final CreatePipePluginPlan createPipePluginPlan) { try { final PipePluginMeta pipePluginMeta = createPipePluginPlan.getPipePluginMeta(); @@ -188,7 +193,7 @@ public TSStatus createPipePlugin(CreatePipePluginPlan createPipePluginPlan) { } return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); - } catch (Exception e) { + } catch (final Exception e) { final String errorMessage = String.format( "Failed to execute createPipePlugin(%s) on config nodes, because of %s", @@ -199,7 +204,7 @@ public TSStatus createPipePlugin(CreatePipePluginPlan createPipePluginPlan) { } } - public TSStatus dropPipePlugin(DropPipePluginPlan dropPipePluginPlan) { + public TSStatus dropPipePlugin(final DropPipePluginPlan dropPipePluginPlan) { final String pluginName = dropPipePluginPlan.getPluginName(); if (pipePluginMetaKeeper.containsPipePlugin(pluginName)) { @@ -217,17 +222,17 @@ public DataSet showPipePlugins() { Arrays.asList(pipePluginMetaKeeper.getAllPipePluginMeta())); } - public JarResp getPipePluginJar(GetPipePluginJarPlan getPipePluginJarPlan) { + public JarResp getPipePluginJar(final GetPipePluginJarPlan getPipePluginJarPlan) { try { final List jarList = new ArrayList<>(); - for (String jarName : getPipePluginJarPlan.getJarNames()) { + for (final String jarName : getPipePluginJarPlan.getJarNames()) { jarList.add( ExecutableManager.transferToBytebuffer( PipePluginExecutableManager.getInstance() .getFileStringUnderInstallByName(jarName))); } return new JarResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()), jarList); - } catch (Exception e) { + } catch (final Exception e) { LOGGER.error("Get PipePlugin_Jar failed", e); return new JarResp( new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()) @@ -239,7 +244,7 @@ public JarResp getPipePluginJar(GetPipePluginJarPlan getPipePluginJarPlan) { /////////////////////////////// Snapshot Processor /////////////////////////////// @Override - public boolean processTakeSnapshot(File snapshotDir) throws IOException { + public boolean processTakeSnapshot(final File snapshotDir) throws IOException { acquirePipePluginInfoLock(); try { final File snapshotFile = new File(snapshotDir, SNAPSHOT_FILE_NAME); @@ -261,7 +266,7 @@ public boolean processTakeSnapshot(File snapshotDir) throws IOException { } @Override - public void processLoadSnapshot(File snapshotDir) throws IOException { + public void processLoadSnapshot(final File snapshotDir) throws IOException { acquirePipePluginInfoLock(); try { final File snapshotFile = new File(snapshotDir, SNAPSHOT_FILE_NAME); @@ -288,7 +293,7 @@ public int hashCode() { } @Override - public boolean equals(Object obj) { + public boolean equals(final Object obj) { if (this == obj) { return true; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java index cb4d6f57fb6c1..1fedccdf8e53c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java @@ -469,7 +469,7 @@ public SettableFuture createFunction( } } catch (IOException | URISyntaxException e) { LOGGER.warn( - "Failed to get executable for UDF({}) using URI: {}, the cause is: {}", + "Failed to get executable for UDF({}) using URI: {}.", createFunctionStatement.getUdfName(), createFunctionStatement.getUriString(), e); @@ -505,14 +505,14 @@ public SettableFuture createFunction( | InvocationTargetException | ClassCastException e) { LOGGER.warn( - "Failed to create function when try to create UDF({}) instance first, the cause is: {}", + "Failed to create function when try to create UDF({}) instance first.", createFunctionStatement.getUdfName(), e); future.setException( new IoTDBException( "Failed to load class '" + createFunctionStatement.getClassName() - + "', because it's not found in jar file: " + + "', because it's not found in jar file or is invalid: " + createFunctionStatement.getUriString(), TSStatusCode.UDF_LOAD_CLASS_ERROR.getStatusCode())); return future; @@ -642,7 +642,7 @@ public SettableFuture createTrigger( } } catch (IOException | URISyntaxException e) { LOGGER.warn( - "Failed to get executable for Trigger({}) using URI: {}, the cause is: {}", + "Failed to get executable for Trigger({}) using URI: {}.", createTriggerStatement.getTriggerName(), createTriggerStatement.getUriString(), e); @@ -679,14 +679,14 @@ public SettableFuture createTrigger( | InvocationTargetException | ClassCastException e) { LOGGER.warn( - "Failed to create trigger when try to create trigger({}) instance first, the cause is: {}", + "Failed to create trigger when try to create trigger({}) instance first.", createTriggerStatement.getTriggerName(), e); future.setException( new IoTDBException( "Failed to load class '" + createTriggerStatement.getClassName() - + "', because it's not found in jar file: " + + "', because it's not found in jar file or is invalid: " + createTriggerStatement.getUriString(), TSStatusCode.TRIGGER_LOAD_CLASS_ERROR.getStatusCode())); return future; @@ -752,7 +752,7 @@ public SettableFuture showTriggers() { @Override public SettableFuture createPipePlugin( - CreatePipePluginStatement createPipePluginStatement) { + final CreatePipePluginStatement createPipePluginStatement) { final SettableFuture future = SettableFuture.create(); final String pluginName = createPipePluginStatement.getPluginName(); final String className = createPipePluginStatement.getClassName(); @@ -768,13 +768,13 @@ public SettableFuture createPipePlugin( try (final ConfigNodeClient client = CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { - String libRoot; - ByteBuffer jarFile; - String jarMd5; + final String libRoot; + final ByteBuffer jarFile; + final String jarMd5; final String jarFileName = new File(uriString).getName(); try { - URI uri = new URI(uriString); + final URI uri = new URI(uriString); if (uri.getScheme() == null) { future.setException( new IoTDBException( @@ -784,10 +784,10 @@ public SettableFuture createPipePlugin( } if (!uri.getScheme().equals("file")) { // Download executable - ExecutableResource resource = + final ExecutableResource resource = PipePluginExecutableManager.getInstance() .request(Collections.singletonList(uriString)); - String jarFilePathUnderTempDir = + final String jarFilePathUnderTempDir = PipePluginExecutableManager.getInstance() .getDirStringUnderTempRootByRequestId(resource.getRequestId()) + File.separator @@ -805,9 +805,9 @@ public SettableFuture createPipePlugin( // Set md5 of the jar file jarMd5 = DigestUtils.md5Hex(Files.newInputStream(Paths.get(libRoot))); } - } catch (IOException | URISyntaxException e) { + } catch (final IOException | URISyntaxException e) { LOGGER.warn( - "Failed to get executable for PipePlugin({}) using URI: {}, the cause is: {}", + "Failed to get executable for PipePlugin({}) using URI: {}.", createPipePluginStatement.getPluginName(), createPipePluginStatement.getUriString(), e); @@ -821,25 +821,26 @@ public SettableFuture createPipePlugin( } // try to create instance, this request will fail if creation is not successful - try (PipePluginClassLoader classLoader = new PipePluginClassLoader(libRoot)) { + try (final PipePluginClassLoader classLoader = new PipePluginClassLoader(libRoot)) { // ensure that jar file contains the class and the class is a pipe plugin - Class clazz = Class.forName(createPipePluginStatement.getClassName(), true, classLoader); - PipePlugin ignored = (PipePlugin) clazz.getDeclaredConstructor().newInstance(); - } catch (ClassNotFoundException + final Class clazz = + Class.forName(createPipePluginStatement.getClassName(), true, classLoader); + final PipePlugin ignored = (PipePlugin) clazz.getDeclaredConstructor().newInstance(); + } catch (final ClassNotFoundException | NoSuchMethodException | InstantiationException | IllegalAccessException | InvocationTargetException | ClassCastException e) { LOGGER.warn( - "Failed to create function when try to create PipePlugin({}) instance first, the cause is: {}", + "Failed to create function when try to create PipePlugin({}) instance first.", createPipePluginStatement.getPluginName(), e); future.setException( new IoTDBException( "Failed to load class '" + createPipePluginStatement.getClassName() - + "', because it's not found in jar file: " + + "', because it's not found in jar file or is invalid: " + createPipePluginStatement.getUriString(), TSStatusCode.PIPE_PLUGIN_LOAD_CLASS_ERROR.getStatusCode())); return future; @@ -868,7 +869,7 @@ public SettableFuture createPipePlugin( } else { future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS)); } - } catch (ClientManagerException | TException | IOException e) { + } catch (final ClientManagerException | TException | IOException e) { future.setException(e); } return future; diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties index 958765e523f09..9e32106da5271 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties @@ -1658,7 +1658,7 @@ data_replication_factor=1 # The total bytes that all pipe sinks can transfer per second. # When given a value less than or equal to 0, it means no limit. # default value is -1, which means no limit. -# Datatype: int +# Datatype: double # pipe_all_sinks_rate_limit_bytes_per_second=-1 #################### diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java index 3d42596e1feee..f72ea5e528d15 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java @@ -562,15 +562,14 @@ private void loadPipeProps(Properties properties) { properties.getProperty( "two_stage_aggregate_sender_end_points_cache_in_ms", String.valueOf(config.getTwoStageAggregateSenderEndPointsCacheInMs())))); + } + private void loadSubscriptionProps(Properties properties) { config.setSubscriptionCacheMemoryUsagePercentage( Float.parseFloat( properties.getProperty( "subscription_cache_memory_usage_percentage", String.valueOf(config.getSubscriptionCacheMemoryUsagePercentage())))); - } - - private void loadSubscriptionProps(Properties properties) { config.setSubscriptionSubtaskExecutorMaxThreadNum( Integer.parseInt( properties.getProperty( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java index ad8e328e45e03..d81292e34d542 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java @@ -117,7 +117,7 @@ public abstract class IoTDBConnector implements PipeConnector { protected PipeReceiverStatusHandler receiverStatusHandler; @Override - public void validate(PipeParameterValidator validator) throws Exception { + public void validate(final PipeParameterValidator validator) throws Exception { final PipeParameters parameters = validator.getParameters(); validator.validate( @@ -232,7 +232,8 @@ public void validate(PipeParameterValidator validator) throws Exception { } @Override - public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration) + public void customize( + final PipeParameters parameters, final PipeConnectorRuntimeConfiguration configuration) throws Exception { nodeUrls.clear(); nodeUrls.addAll(parseNodeUrls(parameters)); @@ -276,7 +277,7 @@ public void customize(PipeParameters parameters, PipeConnectorRuntimeConfigurati CONNECTOR_EXCEPTION_OTHERS_RECORD_IGNORED_DATA_DEFAULT_VALUE)); } - protected LinkedHashSet parseNodeUrls(PipeParameters parameters) + protected LinkedHashSet parseNodeUrls(final PipeParameters parameters) throws PipeParameterNotValidException { final LinkedHashSet givenNodeUrls = new LinkedHashSet<>(nodeUrls); @@ -317,15 +318,22 @@ protected LinkedHashSet parseNodeUrls(PipeParameters parameters) givenNodeUrls.addAll( NodeUrlUtils.parseTEndPointUrls( Arrays.asList( - parameters.getStringByKeys(CONNECTOR_IOTDB_NODE_URLS_KEY).split(",")))); + parameters + .getStringByKeys(CONNECTOR_IOTDB_NODE_URLS_KEY) + .replace(" ", "") + .split(",")))); } if (parameters.hasAttribute(SINK_IOTDB_NODE_URLS_KEY)) { givenNodeUrls.addAll( NodeUrlUtils.parseTEndPointUrls( - Arrays.asList(parameters.getStringByKeys(SINK_IOTDB_NODE_URLS_KEY).split(",")))); + Arrays.asList( + parameters + .getStringByKeys(SINK_IOTDB_NODE_URLS_KEY) + .replace(" ", "") + .split(",")))); } - } catch (Exception e) { + } catch (final Exception e) { LOGGER.warn(PARSE_URL_ERROR_FORMATTER, e.toString()); throw new PipeParameterNotValidException(PARSE_URL_ERROR_MESSAGE); } @@ -335,8 +343,8 @@ protected LinkedHashSet parseNodeUrls(PipeParameters parameters) return givenNodeUrls; } - private void checkNodeUrls(Set nodeUrls) throws PipeParameterNotValidException { - for (TEndPoint nodeUrl : nodeUrls) { + private void checkNodeUrls(final Set nodeUrls) throws PipeParameterNotValidException { + for (final TEndPoint nodeUrl : nodeUrls) { if (Objects.isNull(nodeUrl.ip) || nodeUrl.ip.isEmpty()) { LOGGER.warn(PARSE_URL_ERROR_FORMATTER, "host cannot be empty"); throw new PipeParameterNotValidException(PARSE_URL_ERROR_MESSAGE); @@ -354,13 +362,13 @@ public void close() { PIPE_END_POINT_RATE_LIMITER_MAP.clear(); } - protected TPipeTransferReq compressIfNeeded(TPipeTransferReq req) throws IOException { + protected TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws IOException { return isRpcCompressionEnabled ? PipeTransferCompressedReq.toTPipeTransferReq(req, compressors) : req; } - protected byte[] compressIfNeeded(byte[] reqInBytes) throws IOException { + protected byte[] compressIfNeeded(final byte[] reqInBytes) throws IOException { return isRpcCompressionEnabled ? PipeTransferCompressedReq.toTPipeTransferReqBytes(reqInBytes, compressors) : reqInBytes;