failed to log to fluentd in clojure

Question

To log to fluentd in java API, we need first to create a hashmap and then convert it to map, and then pass the map to the log function like:

Map<String, String> data = new HashMap<String, String>();
        data.put("from", "userA");
        data.put("to", "userB");
        LOG.log("follow", data);

In clojure, I create a hashmap and pass it to log function (but I can't convert hashmap to map, maybe clojure will do that automatically for me), but it failed as below.

The log from fluentd is

2015-05-02 20:52:18 +0800 [warn]: unexpected error error="Could not parse data entirely (0 != 49)"
  2015-05-02 20:52:18 +0800 [warn]: suppressed same stacktrace

The clojure error is

   clojurewerkz.testcom.core=> (def log (org.fluentd.logger.FluentLogger/getLogger "app" "s1" 8888))
    #'clojurewerkz.testcom.core/log
    clojurewerkz.testcom.core=> (def x (java.util.HashMap. {"a" "testa"}))
    #'clojurewerkz.testcom.core/x
    clojurewerkz.testcom.core=> (.log log "test" x)
    true
    clojurewerkz.testcom.core=> (.log log "test" x)
    416694 [nREPL-worker-3] ERROR org.fluentd.logger.sender.RawSocketSender  - org.fluentd.logger.sender.RawSocketSender
    java.net.SocketException: Broken pipe
        at java.net.SocketOutputStream.socketWrite0(Native Method)
        at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:109)
        at java.net.SocketOutputStream.write(SocketOutputStream.java:153)
        at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
        at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
        at org.fluentd.logger.sender.RawSocketSender.flush(RawSocketSender.java:184)
        at org.fluentd.logger.sender.RawSocketSender.send(RawSocketSender.java:172)
        at org.fluentd.logger.sender.RawSocketSender.emit(RawSocketSender.java:142)
        at org.fluentd.logger.sender.RawSocketSender.emit(RawSocketSender.java:124)
        at org.fluentd.logger.sender.RawSocketSender.emit(RawSocketSender.java:119)
        at org.fluentd.logger.FluentLogger.log(FluentLogger.java:100)
        at org.fluentd.logger.FluentLogger.log(FluentLogger.java:85)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:483)
        at clojure.lang.Reflector.invokeMatchingMethod(Reflector.java:93)
        at clojure.lang.Reflector.invokeInstanceMethod(Reflector.java:28)
        at clojurewerkz.testcom.core$eval3467.invoke(form-init3780157560314675153.clj:1)
        at clojure.lang.Compiler.eval(Compiler.java:6703)
        at clojure.lang.Compiler.eval(Compiler.java:6666)
        at clojure.core$eval.invoke(core.clj:2927)
        at clojure.main$repl$read_eval_print__6625$fn__6628.invoke(main.clj:239)
        at clojure.main$repl$read_eval_print__6625.invoke(main.clj:239)
        at clojure.main$repl$fn__6634.invoke(main.clj:257)
        at clojure.main$repl.doInvoke(main.clj:257)
        at clojure.lang.RestFn.invoke(RestFn.java:1523)
        at clojure.tools.nrepl.middleware.interruptible_eval$evaluate$fn__599.invoke(interruptible_eval.clj:67)
        at clojure.lang.AFn.applyToHelper(AFn.java:152)
        at clojure.lang.AFn.applyTo(AFn.java:144)
        at clojure.core$apply.invoke(core.clj:624)
        at clojure.core$with_bindings_STAR_.doInvoke(core.clj:1862)
        at clojure.lang.RestFn.invoke(RestFn.java:425)
        at clojure.tools.nrepl.middleware.interruptible_eval$evaluate.invoke(interruptible_eval.clj:51)
        at clojure.tools.nrepl.middleware.interruptible_eval$interruptible_eval$fn__641$fn__644.invoke(interruptible_eval.clj:183)
        at clojure.tools.nrepl.middleware.interruptible_eval$run_next$fn__634.invoke(interruptible_eval.clj:152)
        at clojure.lang.AFn.run(AFn.java:22)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
    true

Answer 1

I guess you have something wrong with your fluentd configuration. You might be sending forward-type inputs to an http source of a fluentd server. Do you have the following configuration in your fluentd.conf?

<source>
  @type http

  port 8888
</source>

If so, you have to replace http with forward in order to send logs using fluent-logger-java.

<source>
  @type forward

  port 8888
</source>