DataManager/Datastream renamed in BinaryLog

Author: davidecaroselli

README.md

@@ -64,7 +64,7 @@ $ ./mmt status
 [Engine: "default"]
     REST API:   running - 8045/translate
     Cluster:    running - port 5016
-    Datastream: running - localhost:9092
+    Binary log: running - localhost:9092
     Database:   running - localhost:9042
 ```
 

cli/mmt/engine.py

@@ -304,8 +304,8 @@ def __init__(self, props=None):
         self.api_port = props['api']['port'] if 'api' in props else None
         self.api_root = props['api']['root'] if 'api' in props and 'root' in props['api'] else None
         self.cluster_port = props['cluster_port']
-        self.datastream_host = props['datastream']['host'] if 'datastream' in props else None
-        self.datastream_port = props['datastream']['port'] if 'datastream' in props else None
+        self.binlog_host = props['binlog']['host'] if 'binlog' in props else None
+        self.binlog_port = props['binlog']['port'] if 'binlog' in props else None
         self.database_host = props['database']['host'] if 'database' in props else None
         self.database_port = props['database']['port'] if 'database' in props else None
         self.embedded_services = [p for p in props['embedded_services']] if 'embedded_services' in props else None
@@ -342,7 +342,7 @@ def __init__(self, engine):
         self._log_file = engine.get_logfile('node', ensure=False)
         self._api = None
 
-    def start(self, api_port=None, cluster_port=None, datastream_port=None,
+    def start(self, api_port=None, cluster_port=None, binlog_port=None,
               db_port=None, leader=None, verbosity=None, remote_debug=False, log_file=None):
         if log_file is not None:
             self._log_file = log_file
@@ -360,9 +360,9 @@ def start(self, api_port=None, cluster_port=None, datastream_port=None,
             args.append('--api-port')
             args.append(str(api_port))
 
-        if datastream_port is not None:
-            args.append('--datastream-port')
-            args.append(str(datastream_port))
+        if binlog_port is not None:
+            args.append('--binlog-port')
+            args.append(str(binlog_port))
 
         if db_port is not None:
             args.append('--db-port')

cli/server.py

@@ -15,12 +15,12 @@ def parse_args_start(argv=None):
     parser.add_argument('--cluster-port', dest='cluster_port', metavar='CLUSTER_PORT',
                         help='the network port used internally by the cluster for communication between '
                              'Cluster nodes. (default is 5016)', default=None, type=int)
-    parser.add_argument('--datastream-port', dest='datastream_port', metavar='DATASTREAM_PORT',
-                        help='the network port used by Datastream, currently implemented with Kafka '
-                             '(default is 9092', default=None, type=int)
+    parser.add_argument('--binlog-port', '--datastream-port', dest='binlog_port', metavar='BINLOG_PORT',
+                        help='the network port used by BinaryLog, currently implemented with Kafka '
+                             '(default is 9092)', default=None, type=int)
     parser.add_argument('--db-port', dest='db_port', metavar='DB_PORT',
                         help='the network port used by the DB, currently implemented with Cassandra '
-                             '(default is 9042', default=None, type=int)
+                             '(default is 9042)', default=None, type=int)
     parser.add_argument('--join-leader', dest='leader', metavar='NODE_IP', default=None,
                         help='use this option to join this node to an existent cluster. '
                              'NODE is the IP of the remote host to connect to.')
@@ -66,7 +66,7 @@ def main_start(argv=None):
         print('Starting engine "%s"...' % engine.name, end='', flush=True)
         node.start(api_port=args.api_port,
                    cluster_port=args.cluster_port,
-                   datastream_port=args.datastream_port,
+                   binlog_port=args.binlog_port,
                    db_port=args.db_port,
                    leader=args.leader,
                    verbosity=args.verbosity,
@@ -135,13 +135,13 @@ def main_status(argv=None):
             if node_running else 'stopped'
         cluster_s = ('running - port %d' % node_state.cluster_port) \
             if node_running else 'stopped'
-        datastream_s = ('running - %s:%d' % (node_state.datastream_host, node_state.datastream_port)) \
+        binlog_s = ('running - %s:%d' % (node_state.binlog_host, node_state.binlog_port)) \
             if node_running else 'stopped'
         database_s = ('running - %s:%d' % (node_state.database_host, node_state.database_port)) \
             if node_running else 'stopped'
 
         print('[Engine: "%s"]' % engine.name)
         print('    REST API:   %s' % rest_api_s)
         print('    Cluster:    %s' % cluster_s)
-        print('    Datastream: %s' % datastream_s)
+        print('    Binary log: %s' % binlog_s)
         print('    Database:   %s' % database_s)

src/backup-model/src/main/java/eu/modernmt/backup/BackupEngine.java

@@ -1,13 +1,12 @@
 package eu.modernmt.backup;
 
 import eu.modernmt.backup.model.BackupMemory;
-import eu.modernmt.cluster.kafka.KafkaDataManager;
-import eu.modernmt.config.AnalyzerConfig;
-import eu.modernmt.config.DataStreamConfig;
+import eu.modernmt.cluster.kafka.KafkaBinaryLog;
+import eu.modernmt.config.BinaryLogConfig;
 import eu.modernmt.config.NodeConfig;
 import eu.modernmt.context.lucene.LuceneAnalyzer;
-import eu.modernmt.data.DataListener;
-import eu.modernmt.data.DataManager;
+import eu.modernmt.data.LogDataListener;
+import eu.modernmt.data.BinaryLog;
 import eu.modernmt.data.HostUnreachableException;
 import eu.modernmt.decoder.neural.memory.lucene.LuceneTranslationMemory;
 import eu.modernmt.hw.NetworkUtils;
@@ -34,23 +33,23 @@ public class BackupEngine {
 
     private LuceneAnalyzer contextAnalyzer = null;
     private LuceneTranslationMemory memory = null;
-    private DataManager dataManager = null;
+    private BinaryLog binlog = null;
     private BackupMemory backupMemory = null;
 
     public BackupEngine(NodeConfig config, File models) {
         this.uuid = UUID.randomUUID().toString();
         this.config = config;
         this.models = models;
 
-        DataStreamConfig dataStreamConfig = config.getDataStreamConfig();
+        BinaryLogConfig binaryLogConfig = config.getBinaryLogConfig();
 
-        String[] hosts = dataStreamConfig.getHosts();
-        boolean localDatastream = hosts.length == 1 && NetworkUtils.isLocalhost(hosts[0]);
-        boolean embeddedDatastream = dataStreamConfig.isEmbedded();
+        String[] hosts = binaryLogConfig.getHosts();
+        boolean localBinaryLog = hosts.length == 1 && NetworkUtils.isLocalhost(hosts[0]);
+        boolean embeddedBinaryLog = binaryLogConfig.isEmbedded();
 
-        if (embeddedDatastream && localDatastream) {
+        if (embeddedBinaryLog && localBinaryLog) {
             String host = NetworkUtils.getMyIpv4Address();
-            dataStreamConfig.setHost(host);
+            binaryLogConfig.setHost(host);
         }
     }
 
@@ -68,15 +67,15 @@ public void start() throws IOException {
         memory = new LuceneTranslationMemory(Paths.join(models, "memory"), 1);
         backupMemory = new BackupMemory(Paths.join(models, "backup"));
 
-        dataManager = new KafkaDataManager(config.getEngineConfig().getLanguageIndex(), new Preprocessor(), null, uuid, config.getDataStreamConfig());
-        for (DataListener listener : contextAnalyzer.getDataListeners())
-            dataManager.addDataListener(listener);
-        dataManager.addDataListener(memory);
-        dataManager.addDataListener(backupMemory);
+        binlog = new KafkaBinaryLog(config.getEngineConfig().getLanguageIndex(), new Preprocessor(), null, uuid, config.getBinaryLogConfig());
+        for (LogDataListener listener : contextAnalyzer.getDataListeners())
+            binlog.addLogDataListener(listener);
+        binlog.addLogDataListener(memory);
+        binlog.addLogDataListener(backupMemory);
 
         Map<Short, Long> positions;
         try {
-            positions = dataManager.connect(60, TimeUnit.SECONDS, true, false);
+            positions = binlog.connect(60, TimeUnit.SECONDS, true, false);
         } catch (HostUnreachableException e) {
             throw new IOException(e);
         }
@@ -90,7 +89,7 @@ public void stop() throws IOException {
     }
 
     public void stop(boolean optimize) throws IOException {
-        IOException dmError = close(dataManager);
+        IOException dmError = close(binlog);
         IOException caError;
         IOException mError;
         IOException sError;
@@ -132,7 +131,7 @@ public void stop(boolean optimize) throws IOException {
         if (sError != null)
             throw sError;
 
-        dataManager = null;
+        binlog = null;
         contextAnalyzer = null;
         memory = null;
         backupMemory = null;

src/backup-model/src/main/java/eu/modernmt/backup/model/BackupMemory.java

@@ -20,7 +20,7 @@
 import java.util.Map;
 import java.util.function.Consumer;
 
-public class BackupMemory implements Closeable, DataListener {
+public class BackupMemory implements Closeable, LogDataListener {
 
     private final Logger logger = LogManager.getLogger(BackupMemory.class);
 

src/command-line/src/main/java/eu/modernmt/cli/ClusterNodeMain.java

@@ -54,7 +54,7 @@ private static class Args {
 
             Option apiPort = Option.builder("a").longOpt("api-port").hasArg().type(Integer.class).required(false).build();
             Option clusterPort = Option.builder("p").longOpt("cluster-port").hasArg().type(Integer.class).required(false).build();
-            Option datastreamPort = Option.builder().longOpt("datastream-port").hasArg().required(false).build();
+            Option binlogPort = Option.builder().longOpt("binlog-port").hasArg().required(false).build();
             Option databasePort = Option.builder().longOpt("db-port").hasArg().required(false).build();
 
             Option leader = Option.builder().longOpt("leader").hasArg().required(false).build();
@@ -69,7 +69,7 @@ private static class Args {
             cliOptions.addOption(verbosity);
             cliOptions.addOption(leader);
             cliOptions.addOption(logFile);
-            cliOptions.addOption(datastreamPort);
+            cliOptions.addOption(binlogPort);
             cliOptions.addOption(databasePort);
         }
 
@@ -106,13 +106,13 @@ public Args(String[] args) throws ParseException, ConfigException {
             this.verbosity = verbosity == null ? 1 : Integer.parseInt(verbosity);
 
             /* read the engine.xconf file and build the engineConfig, with its:
-               network config, datastream config, database config, net-api config and net-join config */
+               network config, binary-log config, database config, net-api config and net-join config */
             this.config = XMLConfigBuilder.build(Engine.getConfigFile(this.engine));
             this.config.getEngineConfig().setName(this.engine);
 
             // Create the config objects based on the engine.xconf file
             NetworkConfig netConfig = this.config.getNetworkConfig();
-            DataStreamConfig streamConfig = this.config.getDataStreamConfig();
+            BinaryLogConfig binlogConfig = this.config.getBinaryLogConfig();
             DatabaseConfig dbConfig = this.config.getDatabaseConfig();
             ApiConfig apiConfig = netConfig.getApiConfig();
             JoinConfig joinConfig = netConfig.getJoinConfig();
@@ -128,9 +128,9 @@ public Args(String[] args) throws ParseException, ConfigException {
             String apiPort = cli.getOptionValue("api-port");
             if (apiPort != null)
                 apiConfig.setPort(Integer.parseInt(apiPort));
-            String datastreamPort = cli.getOptionValue("datastream-port");
-            if (datastreamPort != null)
-                streamConfig.setPort(Integer.parseInt(datastreamPort));
+            String binlogPort = cli.getOptionValue("binlog-port");
+            if (binlogPort != null)
+                binlogConfig.setPort(Integer.parseInt(binlogPort));
             String databasePort = cli.getOptionValue("db-port");
             if (databasePort != null)
                 dbConfig.setPort(Integer.parseInt(databasePort));
@@ -141,21 +141,21 @@ public Args(String[] args) throws ParseException, ConfigException {
 
             If a leader was passed by command line with option --join-leader:
                 - create a Member with its host and port try to use this Member only to join the cluster
-                - the leader hosts the database and datastream, so set it as the host in datastreamconfig and databaseconfig
+                - the leader hosts the database and binlog, so set it as the host in binaryLogConfig and databaseConfig
 
-            Assume that all the ports that the leader uses (cluster, datastream and cassandra ports) are the same as this node.
+            Assume that all the ports that the leader uses (cluster, binlog and cassandra ports) are the same as this node.
             If they are not, then the configuration is wrong.
 
             If no leader is passed by command line, do nothing.
                 - The config object will use the list of members parsed from engine.xconf and try to join them in order.
-                - use the database and datastream hosts and ports in engine.xconf for datastreamconfig and databaseconfig */
+                - use the database and binlog hosts and ports in engine.xconf for binaryLogConfig and databaseConfig */
             String leader = cli.getOptionValue("leader");
 
             if (leader != null) {
                 JoinConfig.Member[] members = new JoinConfig.Member[1];
                 members[0] = new JoinConfig.Member(leader, netConfig.getPort());
                 joinConfig.setMembers(members);
-                streamConfig.setHost(leader);
+                binlogConfig.setHost(leader);
                 dbConfig.setHost(leader);
             }
         }
@@ -207,7 +207,7 @@ private static class FileStatusListener implements ClusterNode.StatusListener {
          * - cluster port
          * - api config (root and port),
          * - database (host and port)
-         * - datastream (host and port)
+         * - binlog (host and port)
          *
          * @param file   the node.properties JSON file
          *               in which to store the definitive configuration
@@ -219,7 +219,7 @@ public FileStatusListener(File file, NodeConfig config) {
             NetworkConfig netConfig = config.getNetworkConfig();
             ApiConfig apiConfig = netConfig.getApiConfig();
             DatabaseConfig dbConfig = config.getDatabaseConfig();
-            DataStreamConfig streamConfig = config.getDataStreamConfig();
+            BinaryLogConfig binlogConfig = config.getBinaryLogConfig();
 
             this.properties = new JsonObject();
 
@@ -239,11 +239,11 @@ public FileStatusListener(File file, NodeConfig config) {
                 this.properties.add("database", db);
             }
 
-            if (streamConfig.isEnabled()) {
-                JsonObject stream = new JsonObject();
-                stream.addProperty("port", streamConfig.getPort());
-                stream.addProperty("host", StringUtils.join(streamConfig.getHosts(), ','));
-                this.properties.add("datastream", stream);
+            if (binlogConfig.isEnabled()) {
+                JsonObject binlog = new JsonObject();
+                binlog.addProperty("port", binlogConfig.getPort());
+                binlog.addProperty("host", StringUtils.join(binlogConfig.getHosts(), ','));
+                this.properties.add("binlog", binlog);
             }
 
             this.properties.addProperty("cluster_port", netConfig.getPort());

src/commons/src/main/java/eu/modernmt/config/DataStreamConfig.java src/commons/src/main/java/eu/modernmt/config/BinaryLogConfig.java

@@ -6,11 +6,11 @@
  * Created by davide on 04/01/17.
  * Updated by andrearossi on 03/04/17
  * <p>
- * This class embodies a configuration for a DataStream.
+ * This class embodies a configuration for a BinaryLog.
  * It may be read from a configuration file (e.g. engineConf.xml)
  * (or obtained in some different way)
  */
-public class DataStreamConfig {
+public class BinaryLogConfig {
 
     private final NodeConfig parent;
     private boolean enabled = true;
@@ -19,7 +19,7 @@ public class DataStreamConfig {
     private int port = 9092;
     private String name = null;
 
-    public DataStreamConfig(NodeConfig parent) {
+    public BinaryLogConfig(NodeConfig parent) {
         this.parent = parent;
     }
 
@@ -31,7 +31,7 @@ public boolean isEnabled() {
         return enabled;
     }
 
-    public DataStreamConfig setEnabled(boolean enabled) {
+    public BinaryLogConfig setEnabled(boolean enabled) {
         this.enabled = enabled;
         return this;
     }
@@ -40,7 +40,7 @@ public boolean isEmbedded() {
         return this.embedded;
     }
 
-    public DataStreamConfig setEmbedded(boolean embedded) {
+    public BinaryLogConfig setEmbedded(boolean embedded) {
         this.embedded = embedded;
         return this;
     }
@@ -49,7 +49,7 @@ public int getPort() {
         return this.port;
     }
 
-    public DataStreamConfig setPort(int port) {
+    public BinaryLogConfig setPort(int port) {
         this.port = port;
         return this;
     }
@@ -58,12 +58,12 @@ public String[] getHosts() {
         return this.hosts;
     }
 
-    public DataStreamConfig setHost(String host) {
+    public BinaryLogConfig setHost(String host) {
         this.hosts = new String[]{host};
         return this;
     }
 
-    public DataStreamConfig setHosts(String[] hosts) {
+    public BinaryLogConfig setHosts(String[] hosts) {
         this.hosts = hosts;
         return this;
     }
@@ -72,14 +72,14 @@ public String getName() {
         return name;
     }
 
-    public DataStreamConfig setName(String name) {
+    public BinaryLogConfig setName(String name) {
         this.name = name;
         return this;
     }
 
     @Override
     public String toString() {
-        return "Datastream: " +
+        return "Binlog: " +
                 "enabled=" + enabled +
                 ", embedded=" + embedded +
                 ", hosts=" + StringUtils.join(hosts, ',') +