This is an automated email from the ASF dual-hosted git repository.
lta pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5c15ac3 Organize properties (#189)
5c15ac3 is described below
commit 5c15ac3b7bd4ed37e4c91ef2ba3e578fa94bae2b
Author: Tianan Li <[email protected]>
AuthorDate: Wed Jun 5 10:24:29 2019 +0800
Organize properties (#189)
* organize properties
* add time unit
---
iotdb/iotdb/conf/iotdb-engine.properties | 68 ++++++++-----
iotdb/iotdb/conf/iotdb-sync-client.properties | 4 +-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 10 +-
.../org/apache/iotdb/db/sync/conf/Constans.java | 10 ++
.../iotdb/db/sync/conf/SyncSenderConfig.java | 24 +++--
.../iotdb/db/sync/conf/SyncSenderDescriptor.java | 7 +-
.../apache/iotdb/db/sync/sender/SyncSender.java | 10 ++
.../iotdb/db/sync/sender/SyncSenderImpl.java | 106 ++++++++++-----------
8 files changed, 143 insertions(+), 96 deletions(-)
diff --git a/iotdb/iotdb/conf/iotdb-engine.properties
b/iotdb/iotdb/conf/iotdb-engine.properties
index c91cec7..848696c 100644
--- a/iotdb/iotdb/conf/iotdb-engine.properties
+++ b/iotdb/iotdb/conf/iotdb-engine.properties
@@ -17,15 +17,22 @@
# under the License.
#
+####################
+### RPC Configuration
+####################
+
rpc_address=0.0.0.0
rpc_port=6667
-# Write ahead log configuration
+####################
+### Write Ahead Log Configuration
+####################
+
# Is write ahead log enable
enable_wal=true
-# When a certain amount ofwrite ahead log is reached, it will be flushed to
disk
+# When a certain amount of write ahead log is reached, it will be flushed to
disk
# It is possible to lose at most flush_wal_threshold operations
flush_wal_threshold=10000
@@ -38,7 +45,10 @@ flush_wal_period_in_ms=10
# Set this parameter to 0 may slow down the ingestion on slow disk.
force_wal_period_in_ms=10
-# database features configuration
+####################
+### Directory Configuration
+####################
+
# data dir
# If this property is unset, system will save the data in the default relative
path directory under the IoTDB folder(i.e., %IOTDB_HOME%/data/data).
# If it is absolute, system will save the data in exact location it points to.
@@ -100,6 +110,10 @@ force_wal_period_in_ms=10
# If its prefix is "/", then the path is absolute. Otherwise, it is relative.
# wal_dir=/path/iotdb/data
+####################
+### Memory Control Configuration
+####################
+
# The maximum concurrent thread number for merging overflow
# Increase this value, it will increase IO and CPU consumption
# Decrease this value, when there is much overflow data, it will increase disk
usage, which will reduce read speed
@@ -116,7 +130,7 @@ max_opened_folder=100
# The amount of data that is read every time when IoTDB merge data.
fetch_size=10000
-# The period time of flushing data from memory to file.
+# The period time of flushing data from memory to file.
# The unit is second.
period_time_for_flush_in_second=3600
@@ -124,9 +138,9 @@ period_time_for_flush_in_second=3600
# The unit is second.
period_time_for_merge_in_second=7200
-#When set to true, start timing flush and merge service. False, stop timing
flush and merge service.
-#Default is true.
-enable_timing_close_and_Merge=true
+# When set to true, start timing flush and merge service. False, stop timing
flush and merge service.
+# Default is true.
+enable_timing_close_and_merge=true
# time zone of server side
# default value is +08:00
@@ -140,8 +154,8 @@ mem_threshold_warning=0.5
mem_threshold_dangerous=0.6
# every such interval, a thread will check if memory exceeds
mem_threshold_warning
-# if do exceed, auto flush will be triggered, in ms, 1s by default
-mem_monitor_interval=1000
+# if do exceed, auto flush will be triggered, the time unit is ms, 1s by
default
+mem_monitor_interval_in_ms=1000
# Decide how to control memory used by inserting data.
# 0 is RecordMemController, which count the size of every record (tuple).
@@ -164,7 +178,10 @@ overflow_file_size_threshold=209715200
# How many thread can concurrently flush. When <= 0, use CPU core number.
concurrent_flush_thread=0
-# Statistics Monitor configuration
+####################
+### Statistics Monitor configuration
+####################
+
# Set enable_stat_monitor true(or false) to enable(or disable) the StatMonitor
that stores statistics info periodically.
# back_loop_period_sec decides the period when StatMonitor writes statistics
info into IoTDB.
# stat_monitor_detect_freq_sec decides when IoTDB detects statistics info
out-of-date.
@@ -173,14 +190,16 @@ concurrent_flush_thread=0
# The monitor, which writes statistics info to IoTDB periodically, is disabled
by default.
enable_stat_monitor=false
-# The period that StatMonitor stores statistics info
-back_loop_period_sec=5
+# The period that StatMonitor stores statistics info, the time unit is seconds.
+back_loop_period_in_second=5
-# The interval at which StatMonitor starts to check whether statistics info
can be deleted due to exceeding the retention volume
-stat_monitor_detect_freq_sec=600
+# The interval at which StatMonitor starts to check whether statistics info
can be deleted due to exceeding the retention volume.
+# The time unit is seconds.
+stat_monitor_detect_freq_in_second=600
-# The minimum age of statistics storage information to be eligible for
deletion due to age
-stat_monitor_retain_interval_sec=600
+# The minimum age of statistics storage information to be eligible for
deletion due to age.
+# The time unit is seconds.
+stat_monitor_retain_interval_in_second=600
# When set false, MemMonitorThread and MemStatisticThread will not be created.
enable_mem_monitor=true
@@ -202,11 +221,14 @@ schema_manager_cache_size=300000
# Generally the default value 4MB is enough.
max_log_entry_size=4194304
-# IoTDB sync server properties
-# Whether to allow to post back, the default allowed
+####################
+### Sync Server Configuration
+####################
+
+# Whether to open the sync_server_port for receiving data from sync client,
the default allowed
is_sync_enable=true
-# Sync server port address
+# Sync server port to listen
sync_server_port=5555
# White IP list of Sync client.
@@ -215,7 +237,9 @@ sync_server_port=5555
# The default is to allow all IP to sync
IP_white_list=0.0.0.0/0
-# Choose a sync strategy of loading historical data:
-#1. It's more likely to update historical data, please choose "true".
-#2. It's more likely not to update historical data or you don't know exactly,
please choose "false".
+# The processing strategy chosen by the sync server when merging the sync data.
+# 1. If the sync data accounts for more than 50% of the update of the
historical data (compared with the latest timestamp of the local storage group
data),then it is recommended to select strategy 1.
+# Setting the parameter to true, which has a greater impact on the write
performance of the IoTDB system and occupies less CPU of the machine.
+# 2. If the sync data accounts for less than 50% of the update of the
historical data (compared with the latest timestamp of the local storage group
data),then it is recommended to select strategy 2.
+# Setting the parameter to false, which has little impact on the write
performance of IoTDB system and takes up a large amount of CPU power.
update_historical_data_possibility=false
diff --git a/iotdb/iotdb/conf/iotdb-sync-client.properties
b/iotdb/iotdb/conf/iotdb-sync-client.properties
index 479d1c0..65b3074 100644
--- a/iotdb/iotdb/conf/iotdb-sync-client.properties
+++ b/iotdb/iotdb/conf/iotdb-sync-client.properties
@@ -23,8 +23,8 @@ server_ip=127.0.0.1
# Sync client port
server_port=5555
-# The cycle time of post data back to receiver, the unit of time is second
-upload_cycle_in_seconds=600
+# The period time of sync process, the time unit is second.
+sync_period_in_second=600
# Set bufferWrite data absolute path of IoTDB
# It needs to be set with iotdb_schema_directory, they have to belong to the
same IoTDB
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 38892fd..57eedfe 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -87,13 +87,13 @@ public class IoTDBDescriptor {
.parseBoolean(properties.getProperty("enable_stat_monitor",
Boolean.toString(conf.isEnableStatMonitor()))));
conf.setBackLoopPeriodSec(Integer
- .parseInt(properties.getProperty("back_loop_period_sec",
+ .parseInt(properties.getProperty("back_loop_period_in_second",
Integer.toString(conf.getBackLoopPeriodSec()))));
int statMonitorDetectFreqSec = Integer.parseInt(
- properties.getProperty("stat_monitor_detect_freq_sec",
+ properties.getProperty("stat_monitor_detect_freq_in_second",
Integer.toString(conf.getStatMonitorDetectFreqSec())));
int statMonitorRetainIntervalSec = Integer.parseInt(
- properties.getProperty("stat_monitor_retain_interval_sec",
+ properties.getProperty("stat_monitor_retain_interval_in_second",
Integer.toString(conf.getStatMonitorRetainIntervalSec())));
// the conf value must > default value, or may cause system unstable
if (conf.getStatMonitorDetectFreqSec() < statMonitorDetectFreqSec) {
@@ -157,7 +157,7 @@ public class IoTDBDescriptor {
properties.getProperty("period_time_for_merge_in_second",
Long.toString(conf.getPeriodTimeForMerge())).trim()));
conf.setEnableTimingCloseAndMerge(Boolean.parseBoolean(properties
- .getProperty("enable_timing_close_and_Merge",
+ .getProperty("enable_timing_close_and_merge",
Boolean.toString(conf.isEnableTimingCloseAndMerge())).trim()));
conf.setMemThresholdWarning((long) (Runtime.getRuntime().maxMemory() *
Double.parseDouble(
@@ -168,7 +168,7 @@ public class IoTDBDescriptor {
Long.toString(conf.getMemThresholdDangerous())).trim())));
conf.setMemMonitorInterval(Long
- .parseLong(properties.getProperty("mem_monitor_interval",
+ .parseLong(properties.getProperty("mem_monitor_interval_in_ms",
Long.toString(conf.getMemMonitorInterval())).trim()));
conf.setMemControllerType(Integer
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
index f43207f..c810c4c 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
@@ -44,4 +44,14 @@ public class Constans {
*/
public static final int MAX_SYNC_FILE_TRY = 10;
+ private static final SyncSenderConfig CONFIG =
SyncSenderDescriptor.getInstance().getConfig();
+
+ public static final long SYNC_PROCESS_DELAY = 0;
+
+ public static final long SYNC_MONITOR_DELAY = CONFIG.getSyncPeriodInSecond();
+
+ public static final long SYNC_PROCESS_PERIOD =
CONFIG.getSyncPeriodInSecond();
+
+ public static final long SYNC_MONITOR_PERIOD =
CONFIG.getSyncPeriodInSecond();
+
}
diff --git
a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
index 50bd443..727cde3 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
@@ -22,23 +22,33 @@ import java.io.File;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.metadata.MetadataConstant;
import org.apache.iotdb.db.utils.FilePathUtils;
-import org.apache.iotdb.db.utils.FileUtils;
-import org.apache.iotdb.db.utils.SyncUtils;
public class SyncSenderConfig {
private String[] bufferwriteDirectory =
IoTDBDescriptor.getInstance().getConfig()
.getBufferWriteDirs();
+
private String dataDirectory =
IoTDBDescriptor.getInstance().getConfig().getDataDir();
+
private String lockFilePath;
+
private String uuidPath;
+
private String lastFileInfo;
+
private String[] snapshotPaths;
+
private String schemaPath;
+
private String serverIp = "127.0.0.1";
+
private int serverPort = 5555;
- private int uploadCycleInSeconds = 10;
+ private int syncPeriodInSecond = 10;
+
+ /**
+ * Init path
+ */
public void init() {
String metadataDirPath =
IoTDBDescriptor.getInstance().getConfig().getMetadataDir();
metadataDirPath = FilePathUtils.regularizePath(metadataDirPath);
@@ -127,12 +137,12 @@ public class SyncSenderConfig {
this.serverPort = serverPort;
}
- public int getUploadCycleInSeconds() {
- return uploadCycleInSeconds;
+ public int getSyncPeriodInSecond() {
+ return syncPeriodInSecond;
}
- public void setUploadCycleInSeconds(int uploadCycleInSeconds) {
- this.uploadCycleInSeconds = uploadCycleInSeconds;
+ public void setSyncPeriodInSecond(int syncPeriodInSecond) {
+ this.syncPeriodInSecond = syncPeriodInSecond;
}
public String getLockFilePath() {
diff --git
a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
index 99dcb56..ec1dc03 100644
---
a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
+++
b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
@@ -26,7 +26,6 @@ import java.io.InputStream;
import java.util.Properties;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.utils.FilePathUtils;
-import org.apache.iotdb.db.utils.SyncUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -88,9 +87,9 @@ public class SyncSenderDescriptor {
conf.setServerIp(properties.getProperty("server_ip",
conf.getServerIp()));
conf.setServerPort(Integer
.parseInt(properties.getProperty("server_port",
Integer.toString(conf.getServerPort()))));
- conf.setUploadCycleInSeconds(Integer.parseInt(properties
- .getProperty("upload_cycle_in_seconds",
- Integer.toString(conf.getUploadCycleInSeconds()))));
+ conf.setSyncPeriodInSecond(Integer.parseInt(properties
+ .getProperty("sync_period_in_second",
+ Integer.toString(conf.getSyncPeriodInSecond()))));
conf.setSchemaPath(properties.getProperty("iotdb_schema_directory",
conf.getSchemaPath()));
conf.setDataDirectory(
properties.getProperty("iotdb_bufferWrite_directory",
conf.getDataDirectory()));
diff --git
a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/SyncSender.java
b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/SyncSender.java
index 4ef1a3c..d45a09e 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/SyncSender.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/SyncSender.java
@@ -28,6 +28,11 @@ import org.apache.iotdb.db.exception.SyncConnectionException;
public interface SyncSender {
/**
+ * Init
+ */
+ void init();
+
+ /**
* Connect to server.
*/
void establishConnection(String serverIp, int serverPort) throws
SyncConnectionException;
@@ -62,4 +67,9 @@ public interface SyncSender {
*/
void sync() throws SyncConnectionException, IOException;
+ /**
+ * Stop sync process
+ */
+ void stop();
+
}
diff --git
a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/SyncSenderImpl.java
b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/SyncSenderImpl.java
index a20ff7d..0fbfac4 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/SyncSenderImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/SyncSenderImpl.java
@@ -34,7 +34,6 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.util.ArrayList;
-import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -42,7 +41,10 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.db.concurrent.ThreadName;
import org.apache.iotdb.db.exception.SyncConnectionException;
import org.apache.iotdb.db.sync.conf.Constans;
@@ -66,10 +68,15 @@ import org.slf4j.LoggerFactory;
public class SyncSenderImpl implements SyncSender {
private static final Logger LOGGER =
LoggerFactory.getLogger(SyncSenderImpl.class);
+
private TTransport transport;
+
private SyncService.Client serviceClient;
+
private List<String> schema = new ArrayList<>();
+ private static SyncSenderConfig config =
SyncSenderDescriptor.getInstance().getConfig();
+
/**
* Files that need to be synchronized
*/
@@ -81,11 +88,6 @@ public class SyncSenderImpl implements SyncSender {
private Map<String, Set<String>> currentLocalFiles;
/**
- * Mark the start time of last sync
- **/
- private Date lastSyncTime = new Date();
-
- /**
* If true, sync is in execution.
**/
private volatile boolean syncStatus = false;
@@ -96,29 +98,11 @@ public class SyncSenderImpl implements SyncSender {
private Map<String, Set<String>> validFileSnapshot = new HashMap<>();
private SyncFileManager syncFileManager = SyncFileManager.getInstance();
- private SyncSenderConfig config =
SyncSenderDescriptor.getInstance().getConfig();
- /**
- * Monitor sync status.
- */
- private final Runnable monitorSyncStatus = () -> {
- Date oldTime = new Date();
- while (!Thread.interrupted()) {
- Date currentTime = new Date();
- if (currentTime.getTime() / 1000 == oldTime.getTime() / 1000) {
- continue;
- }
- if ((currentTime.getTime() - lastSyncTime.getTime())
- % (config.getUploadCycleInSeconds() * 1000) == 0) {
- oldTime = currentTime;
- if (syncStatus) {
- LOGGER.info("Sync process is in execution!");
- }
- }
- }
- };
+ private ScheduledExecutorService executorService;
private SyncSenderImpl() {
+ init();
}
public static final SyncSenderImpl getInstance() {
@@ -130,43 +114,51 @@ public class SyncSenderImpl implements SyncSender {
*
* @param args not used
*/
- public static void main(String[] args)
- throws InterruptedException, IOException, SyncConnectionException {
+ public static void main(String[] args) throws IOException {
Thread.currentThread().setName(ThreadName.SYNC_CLIENT.getName());
SyncSenderImpl fileSenderImpl = new SyncSenderImpl();
fileSenderImpl.verifySingleton();
fileSenderImpl.startMonitor();
- fileSenderImpl.timedTask();
+ fileSenderImpl.startTimedTask();
+ }
+
+ @Override
+ public void init() {
+ if (executorService == null) {
+ executorService = IoTDBThreadPoolFactory.newScheduledThreadPool(2,
+ "sync-client-timer");
+ }
}
/**
* Start Monitor Thread, monitor sync status
*/
- public void startMonitor() {
- Thread syncMonitor = new Thread(monitorSyncStatus,
ThreadName.SYNC_MONITOR.getName());
- syncMonitor.setDaemon(true);
- syncMonitor.start();
+ private void startMonitor() {
+ executorService.scheduleWithFixedDelay(() -> {
+ if (syncStatus) {
+ LOGGER.info("Sync process is in execution!");
+ }
+ }, Constans.SYNC_MONITOR_DELAY, Constans.SYNC_MONITOR_PERIOD,
TimeUnit.SECONDS);
}
/**
* Start sync task in a certain time.
*/
- public void timedTask() throws InterruptedException,
SyncConnectionException, IOException {
- sync();
- lastSyncTime = new Date();
- Date currentTime;
- while (true) {
- if (Thread.interrupted()) {
- break;
- }
- Thread.sleep(2000);
- currentTime = new Date();
- if (currentTime.getTime() - lastSyncTime.getTime()
- > config.getUploadCycleInSeconds() * 1000) {
- lastSyncTime = currentTime;
+ private void startTimedTask() {
+ executorService.scheduleWithFixedDelay(() -> {
+ try {
sync();
+ } catch (SyncConnectionException | IOException e) {
+ LOGGER.error("Sync failed", e);
+ stop();
}
- }
+ }, Constans.SYNC_PROCESS_DELAY, Constans.SYNC_PROCESS_PERIOD,
TimeUnit.SECONDS);
+ }
+
+ @Override
+ public void stop() {
+ executorService.shutdownNow();
+ executorService = null;
}
/**
@@ -178,7 +170,7 @@ public class SyncSenderImpl implements SyncSender {
//1. Clear old snapshots if necessary
for (String snapshotPath : config.getSnapshotPaths()) {
if (new File(snapshotPath).exists() && new
File(snapshotPath).list().length != 0) {
- /** It means that the last task of sync does not succeed! Clear the
files and start to sync again **/
+ // It means that the last task of sync does not succeed! Clear the
files and start to sync again
FileUtils.deleteDirectory(new File(snapshotPath));
}
}
@@ -206,13 +198,13 @@ public class SyncSenderImpl implements SyncSender {
syncStatus = true;
- try{
+ try {
// 5. Sync schema
syncSchema();
// 6. Sync data
syncAllData();
- }catch (SyncConnectionException e){
+ } catch (SyncConnectionException e) {
LOGGER.error("cannot finish sync process", e);
syncStatus = false;
return;
@@ -245,7 +237,7 @@ public class SyncSenderImpl implements SyncSender {
}
LOGGER.info("Sync process starts to transfer data of storage group {}",
entry.getKey());
try {
- if(!serviceClient.init(entry.getKey())){
+ if (!serviceClient.init(entry.getKey())) {
throw new SyncConnectionException("unable init receiver");
}
} catch (TException e) {
@@ -258,7 +250,8 @@ public class SyncSenderImpl implements SyncSender {
syncFileManager.backupNowLocalFileInfo(config.getLastFileInfo());
LOGGER.info("Sync process has finished storage group {}.",
entry.getKey());
} else {
- LOGGER.error("Receiver cannot sync data, abandon this synchronization
of storage group {}", entry.getKey());
+ LOGGER.error("Receiver cannot sync data, abandon this synchronization
of storage group {}",
+ entry.getKey());
}
}
}
@@ -380,7 +373,7 @@ public class SyncSenderImpl implements SyncSender {
while (true) {
retryCount++;
// Sync all data to receiver
- if(retryCount > Constans.MAX_SYNC_FILE_TRY){
+ if (retryCount > Constans.MAX_SYNC_FILE_TRY) {
throw new SyncConnectionException(String
.format("can not sync file %s after %s tries.",
snapshotFilePath,
Constans.MAX_SYNC_FILE_TRY));
@@ -395,8 +388,8 @@ public class SyncSenderImpl implements SyncSender {
md.update(buffer, 0, dataLength);
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
- if(!Boolean.parseBoolean(serviceClient
- .syncData(null, filePathSplit, buffToSend,
SyncDataStatus.PROCESSING_STATUS))){
+ if (!Boolean.parseBoolean(serviceClient
+ .syncData(null, filePathSplit, buffToSend,
SyncDataStatus.PROCESSING_STATUS))) {
LOGGER.info("Receiver failed to receive data from {}, retry.",
snapshotFilePath);
continue outer;
}
@@ -445,7 +438,8 @@ public class SyncSenderImpl implements SyncSender {
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
// PROCESSING_STATUS represents there is still schema buffer to send.
- if(!Boolean.parseBoolean(serviceClient.syncSchema(null, buffToSend,
SyncDataStatus.PROCESSING_STATUS))){
+ if (!Boolean.parseBoolean(
+ serviceClient.syncSchema(null, buffToSend,
SyncDataStatus.PROCESSING_STATUS))) {
LOGGER.error("Receiver failed to receive metadata, retry.");
continue outer;
}
