• 前提是配置consistency参数为true
• es监听中afterBulk(long arg0, BulkRequest arg1, Throwable arg2)方法：

public void afterBulk(long arg0, BulkRequest arg1, Throwable arg2) {
          logger.error("bulk got exception:", arg2);
                        
          for(ActionRequest request : arg1.requests()){
               addFailedMsg(request);
          }
                        
}

该方法中的addFailedMsg(request)传入的参数为ActionRequest，而 addFailedMsg(Object msg):

public void addFailedMsg(Object msg) {
        if (msg instanceof Map) {
            super.addFailedMsg(msg);
            return;
        }

        throw new IllegalArgumentException("addFailedMsg only accept Map instance");
    }

中实际处理时候做了msg是否为 Map 的判断，所以从这里进去的数据永远不会被添加到 failedMsgQueue，从而导致数据丢失

如果启动2个进程，配置的相同的配置文件，读取的相同文件一共1000 条数据，数据是怎么平均分配到这2个线程，每个500条。看了下源码，基本是多线程和队列都是进程内的，如何保证集群了读取的 数据不重复那？咨询下。谢谢.QQ:907519797, vx：17865621181

大神，刚开始使用jlogstash。感觉使用说明不太详细，接入的时候，感觉很吃力。。弄一个jgrok格式，都弄老半天。

实现类似于下面的功能：

filter {
  if [loglevel] == "debug" {
    drop { }
  }
}

Remove 可以实现吗？如果可以，请帮忙写的例示。非常感谢！

找不到启动类，核心代码目录是空的，此软件是收费的吗?

如果能比较一下与flume的性能差异就更好了。
另外提一句,
代码中有很多不合理的地方，尤其是插件设计的部分

jdbc的input插件中怎样配置增量读取？logstash中已更新时间来做增量的

logstash 官方从5.x 开始 提供将java版本

比如：
Logstash 5.0.0-beta1
https://artifacts.elastic.co/downloads/logstash/logstash-5.0.0-beta1.zip

请问是否支持elasticsearch7.0以上版本，如果es开启了安全验证，这边怎么配置呢？

怎么lib和plugin里面都是空的了

使用kafka10的inputs插件时，报以下error，启动失败。
2019-06-24 17:11:04.218 [main] WARN com.dtstack.jlogstash.factory.InstanceFactory [InstanceFactory.java:64] - field: topics annotation:Required check
2019-06-24 17:11:04.221 [main] ERROR com.dtstack.jlogstash.JlogstashMain [JlogstashMain.java:53] - jlogstash start error:null
com.dtstack.jlogstash.annotation.plugin.RequiredPlugin.required(39)
com.dtstack.jlogstash.factory.InstanceFactory.checkAnnotation(139)
com.dtstack.jlogstash.factory.InstanceFactory.configInstance(65)
com.dtstack.jlogstash.factory.InputFactory.getInstance(44)
com.dtstack.jlogstash.factory.InputFactory.getBatchInstance(64)
com.dtstack.jlogstash.assembly.AssemblyPipeline.assemblyPipeline(96)
com.dtstack.jlogstash.JlogstashMain.main(51)

debug模式调试，定位代码com.dtstack.jlogstash.annotation.plugin.RequiredPlugin中required 为空指针。请看看是什么原因？
public void required(Field field,Object obj) throws Exception{
// TODO Auto-generated method stub
Required required = field.getAnnotation(Required.class);
if(required.required()&&obj==null){
throw new RequiredException(field.getName());
}

配置文件：
inputs:
- Kafka10:
codec: json
encoding: UTF8 # defaut UTF8
topics: "my-slf4j-topic-test"
groupId: "test"
bootstrapServers: 172.18.231.8:32129,172.18.231.8:32130
consumerSettings:
auto.commit.interval.ms: "1000"
outputs:
- File:
path: E:/dockerhub/jlogstash/srsyslog-performance-%{+YYYY.MM.dd}.txt

jlogstash-1.0.0-with-dependencies.jar 核心代码放在jlogstash/lib/下

RT

求发一个可以根据grok 正则表达式正确解析并输出日志文件的配置文件例子

大神你好!!
我抓了你的包後，在lib內所需要的jar是否要自己編寫嗎?

你好
使用JLogstash日志滚动问题要怎么解决

你好,非常喜欢你的项目，想在你项目基础上进行一些开发，请问你项目目前支持单个yml配置文件配置多表吗？如果支持请问是怎么配置的，还有如何在配置文件里配置logstash那样的文档主键id？？？？谢谢

2019-07-02 13:44:10.619 [main] INFO org.elasticsearch.plugins.PluginsService [PluginsService.java:180] - loaded plugin [org.elasticsearch.transport.Netty4Plugin]
java.lang.OutOfMemoryError: Java heap space
Dumping heap to ../logs/heapdump.hprof ...
Heap dump file created [380426013 bytes in 2.713 secs]
Exception in thread "elasticsearch[client][scheduler][T#1]" java.lang.OutOfMemoryError: Java heap space
Exception in thread "elasticsearch[client][scheduler][T#1]" java.lang.IllegalMonitorStateException
2019-07-02 13:45:13.213 [elasticsearch[client][transport_client_boss][T#5]] WARN io.netty.util.ReferenceCountUtil [Netty4InternalESLogger.java:149] - Failed to release a message: PooledDirectByteBuf(freed)
java.lang.OutOfMemoryError: Java heap space
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.signal(AbstractQueuedSynchronizer.java:1939)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1103)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "elasticsearch[client][generic][T#1]"

provided
不去掉的话 反射对象报错，
去掉了 启动的时候又找不到类了,看了一下 包里面是有这个类的
ERROR com.dtstack.jlogstash.inputs.Kafka10 [Kafka10.java:101] - kafka emit error
org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.StringDeserializer for configuration value.deserializer: Class org.apache.kafka.common.serialization.StringDeserializer could not be found.

JLogStash使用文件输入、Netty输出的时候，在日志量很少或者没有的情况下，CPU依然会占用5-8%，管理器上面看到待唤醒线程数有2000+

能否更新使用说明呢，最好有示例，我无法使用大部分的input插件采集到数据，总是提示这样或者那样的问题

看代码中使用的是阻塞队列，Input数据读取完后，OutputThread还被阻塞在，没办法自动退出虚拟机。

OutputThread中部分代码：　
if(!priorityFail()){
event = this.outputQueue.take();
if (event != null) {
for (BaseOutput bo : outputProcessors) {
bo.process(event);
}
}
}

是否可以上传下TestInputPlugin类的emit prepare和release方法 还不知道下面具体的逻辑 还有fliter和 output 或者可以给下思路 该如何继续下去 谢谢哈

初学jlogstash，启动时遇到一些问题，请帮忙看下
操作时上传打成的jar包jlogstash-1.0.0.jar到lib目录下，编写test.yml文件，执行命令报错：
Error: Could not find or load main class .test.jlogstash.lib.logs
test是目录名

单独打包 popeline下inputs 的file项目，出现 Cannot run program "gpg.exe": CreateProcess error=2, The system cannot find the file specified问题 ，这个是需要单独安装gpg.exe 再配置环境变量是么 ？
可否，方便有个开发环境搭建文档

请问filebeat可以直接使用jlogstash吗？

在InputFactor.java中46行，导致类型转换失败，不知道大神遇到过没，怎么解决
jlogstash start error:com.dtstack.jlogstash.inputs.Kafka10 cannot be cast to com.dtstack.jlogstash.inputs.BaseInput
com.dtstack.jlogstash.factory.InputFactory.getInstance(46)
com.dtstack.jlogstash.factory.InputFactory.getBatchInstance(64)
com.dtstack.jlogstash.assembly.AssemblyPipeline.assemblyPipeline(103)
com.dtstack.jlogstash.JlogstashMain.main(51)

看了說明還是不太明白怎麼去設置filters
可以請大神幫忙一下嗎??

已下是我用官方的logstsh 設置的設定檔
input {
beats {
port => 5044
client_inactivity_timeout => 300
}
}

filter{

if "beats_input_codec_plain_applied" in [tags] {
mutate {
remove_tag => ["beats_input_codec_plain_applied"]
}
}
if "_geoip_lookup_failure" in [tags] {
drop { }
}

if "_grokparsefailure" in [tags] {
drop { }
}
if [xclientip] == "-" {
mutate{
replace => { "xclientip" => "0.0.0.0" }
}
}
if [type] in [ "aa", "bb", "cc" ] {

grok{
patterns_dir => ["/usr/local/logstash/patterns"]
match => [ "message", "%{COMBINEDAPACHELOG2}" ]
}
geoip{
source => "xclientip"
target => "geoip"
database => "/usr/local/logstash/GeoIP/GeoLite2-City.mmdb"
add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}
mutate{
convert => [ "[geoip][coordinates]", "float", "bytes", "integer", "elapsedmillis", "integer" ]
}
date {
match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" , "ISO8601" ]
target => "@timestamp"

}

}
}

output {
if [type] in [ "aa", "bb", "cc" ] {
elasticsearch {
hosts =>[ "192.168.1.111:9200","192.168.1.110:9200" ]
index => "%{type}-%{[@metadata][beat]}-%{+YYYY.MM.dd}"

flush_size => 50000

timeout => "60"
document_type => "%{[@metadata][type]}"
template => "/usr/local/logstash/filebeat-index-template.json"
template_overwrite => true
user => 'elastic'
password => 'changeme'

}
}
}