Merge pull request zendesk#73 from zendesk/file_producer_fixes

Ben Osheroff · Ben Osheroff · commit 86c45f4a11bb · 2015-07-08T12:54:40.000-07:00
fixes for file producer
diff --git a/src/main/java/com/zendesk/maxwell/MaxwellConfig.java b/src/main/java/com/zendesk/maxwell/MaxwellConfig.java
@@ -49,6 +49,7 @@ private OptionParser getOptionParser() {
 		parser.accepts( "log_level", "log level, one of DEBUG|INFO|WARN|ERROR" ).withRequiredArg();
 		parser.accepts( "host", "mysql host" ).withRequiredArg();
 		parser.accepts( "user", "mysql username" ).withRequiredArg();
+		parser.accepts( "output_file", "output file for 'file' producer" ).withRequiredArg();
 		parser.accepts( "password", "mysql password" ).withRequiredArg();
 		parser.accepts( "port", "mysql port" ).withRequiredArg();
 		parser.accepts( "producer", "producer type: stdout|file|kafka" ).withRequiredArg();
@@ -97,6 +98,9 @@ private void parse(String [] argv) {
 
 		if ( options.has("kafka_topic"))
 			this.kafkaTopic = (String) options.valueOf("kafka_topic");
+
+		if ( options.has("output_file"))
+			this.outputFile = (String) options.valueOf("output_file");
 	}
 
 	private Properties readPropertiesFile(String filename, Boolean abortOnMissing) {
@@ -155,6 +159,9 @@ private void setDefaults() {
 		} else if ( this.producerType.equals("kafka")
 				&& !this.kafkaProperties.containsKey("bootstrap.servers")) {
 			usage("You must specify kafka.bootstrap.servers for the kafka producer!");
+		} else if ( this.producerType.equals("file")
+				&& this.outputFile == null) {
+			usage("please specify --output_file=FILE to use the file producer");
 		}
 
 		if ( this.mysqlPort == null )
diff --git a/src/main/java/com/zendesk/maxwell/producer/AbstractProducer.java b/src/main/java/com/zendesk/maxwell/producer/AbstractProducer.java
@@ -11,7 +11,5 @@ public AbstractProducer(MaxwellContext context) {
 	}
 	abstract public void push(MaxwellAbstractRowsEvent e) throws Exception;
 
-	public void onComplete(MaxwellAbstractRowsEvent e) {
-		System.out.println("processed " + e.getBinlogFilename() + ":" + e.getHeader().getPosition());
-	}
+	public void onComplete(MaxwellAbstractRowsEvent e) { }
 }
diff --git a/src/main/java/com/zendesk/maxwell/producer/FileProducer.java b/src/main/java/com/zendesk/maxwell/producer/FileProducer.java
@@ -24,7 +24,6 @@ public void push(MaxwellAbstractRowsEvent e) throws Exception {
 			this.fileWriter.write('\n');
 			this.fileWriter.flush();
 		}
-		this.onComplete(e);
 		context.setInitialPosition(e.getNextBinlogPosition());
 	}
 }

Original file line number	Diff line number	Diff line change
`@@ -11,7 +11,5 @@ public AbstractProducer(MaxwellContext context) {`
`11`	`11`	`}`
`12`	`12`	`abstract public void push(MaxwellAbstractRowsEvent e) throws Exception;`
`13`	`13`
`14`		`- public void onComplete(MaxwellAbstractRowsEvent e) {`
`15`		`- System.out.println("processed " + e.getBinlogFilename() + ":" + e.getHeader().getPosition());`
`16`		`- }`
	`14`	`+ public void onComplete(MaxwellAbstractRowsEvent e) { }`
`17`	`15`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,6 @@ public void push(MaxwellAbstractRowsEvent e) throws Exception {`
`24`	`24`	`this.fileWriter.write('\n');`
`25`	`25`	`this.fileWriter.flush();`
`26`	`26`	`}`
`27`		`- this.onComplete(e);`
`28`	`27`	`context.setInitialPosition(e.getNextBinlogPosition());`
`29`	`28`	`}`
`30`	`29`	`}`