Hi Koji, sorry about the provenance exception, it was because there's no space left on the machine(filled up with logs)
Regards, Ben 2017-12-27 17:11 GMT+08:00 尹文才 <batman...@gmail.com>: > Hi Koji, thanks, the names of the temp tables are created with format > "yyyyMMddHHmmssSSS-nnnn", the first time indicates the time and the second > part is a random number with length of 4. > So I think it's not possible to have 2 duplicate table names, the only > possibly I could think is the flowfile is passed into the processor twice. > > About the provenance, I had updated to use the > WriteAheadProvenanceRepository implementation, but when I tried to check > the data provenance, it showed me the following exception message: > HTTP ERROR 500 > > Problem accessing /nifi/provenance. Reason: > > Server Error > > Caused by: > > javax.servlet.ServletException: org.eclipse.jetty.servlet.ServletHolder$1: > java.lang.NullPointerException > at > org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:138) > at > org.eclipse.jetty.server.handler.gzip.GzipHandler.handle(GzipHandler.java:561) > at > org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:132) > at org.eclipse.jetty.server.Server.handle(Server.java:564) > at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:320) > at > org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:251) > at > org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:279) > at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:110) > at > org.eclipse.jetty.io.ssl.SslConnection.onFillable(SslConnection.java:258) > at > org.eclipse.jetty.io.ssl.SslConnection$3.succeeded(SslConnection.java:147) > at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:110) > at org.eclipse.jetty.io.ChannelEndPoint$2.run(ChannelEndPoint.java:124) > at > org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:672) > at > org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:590) > at java.lang.Thread.run(Thread.java:745) > Caused by: org.eclipse.jetty.servlet.ServletHolder$1: > java.lang.NullPointerException > at > org.eclipse.jetty.servlet.ServletHolder.makeUnavailable(ServletHolder.java:596) > at > org.eclipse.jetty.servlet.ServletHolder.initServlet(ServletHolder.java:655) > at > org.eclipse.jetty.servlet.ServletHolder.getServlet(ServletHolder.java:498) > at > org.eclipse.jetty.servlet.ServletHolder.ensureInstance(ServletHolder.java:785) > at > org.eclipse.jetty.servlet.ServletHolder.prepare(ServletHolder.java:770) > at > org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:538) > at > org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:143) > at > org.eclipse.jetty.security.SecurityHandler.handle(SecurityHandler.java:548) > at > org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:132) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:190) > at > org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1593) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:188) > at > org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1239) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:168) > at > org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:481) > at > org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1562) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:166) > at > org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1141) > at > org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141) > at > org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:118) > at > org.eclipse.jetty.server.handler.gzip.GzipHandler.handle(GzipHandler.java:561) > at > org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:132) > at org.eclipse.jetty.server.Server.handle(Server.java:564) > at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:320) > at > org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:251) > at > org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:279) > at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:110) > at > org.eclipse.jetty.io.ssl.SslConnection.onFillable(SslConnection.java:258) > at > org.eclipse.jetty.io.ssl.SslConnection$3.succeeded(SslConnection.java:147) > at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:110) > at org.eclipse.jetty.io.ChannelEndPoint$2.run(ChannelEndPoint.java:124) > at > org.eclipse.jetty.util.thread.Invocable.invokePreferred(Invocable.java:122) > at > org.eclipse.jetty.util.thread.strategy.ExecutingExecutionStrategy.invoke(ExecutingExecutionStrategy.java:58) > at > org.eclipse.jetty.util.thread.strategy.ExecuteProduceConsume.produceConsume(ExecuteProduceConsume.java:201) > at > org.eclipse.jetty.util.thread.strategy.ExecuteProduceConsume.run(ExecuteProduceConsume.java:133) > ... 3 more > Caused by: java.lang.NullPointerException > at > org.apache.jasper.servlet.JspServlet.handleMissingResource(JspServlet.java:397) > at > org.apache.jasper.servlet.JspServlet.serviceJspFile(JspServlet.java:387) > at org.apache.jasper.servlet.JspServlet.init(JspServlet.java:138) > at > org.eclipse.jetty.servlet.ServletHolder.initServlet(ServletHolder.java:637) > ... 36 more > > Caused by: > > org.eclipse.jetty.servlet.ServletHolder$1: java.lang.NullPointerException > at > org.eclipse.jetty.servlet.ServletHolder.makeUnavailable(ServletHolder.java:596) > at > org.eclipse.jetty.servlet.ServletHolder.initServlet(ServletHolder.java:655) > at > org.eclipse.jetty.servlet.ServletHolder.getServlet(ServletHolder.java:498) > at > org.eclipse.jetty.servlet.ServletHolder.ensureInstance(ServletHolder.java:785) > at > org.eclipse.jetty.servlet.ServletHolder.prepare(ServletHolder.java:770) > at > org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:538) > at > org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:143) > at > org.eclipse.jetty.security.SecurityHandler.handle(SecurityHandler.java:548) > at > org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:132) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:190) > at > org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1593) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:188) > at > org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1239) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:168) > at > org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:481) > at > org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1562) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:166) > at > org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1141) > at > org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141) > at > org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:118) > at > org.eclipse.jetty.server.handler.gzip.GzipHandler.handle(GzipHandler.java:561) > at > org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:132) > at org.eclipse.jetty.server.Server.handle(Server.java:564) > at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:320) > at > org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:251) > at > org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:279) > at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:110) > at > org.eclipse.jetty.io.ssl.SslConnection.onFillable(SslConnection.java:258) > at > org.eclipse.jetty.io.ssl.SslConnection$3.succeeded(SslConnection.java:147) > at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:110) > at org.eclipse.jetty.io.ChannelEndPoint$2.run(ChannelEndPoint.java:124) > at > org.eclipse.jetty.util.thread.Invocable.invokePreferred(Invocable.java:122) > at > org.eclipse.jetty.util.thread.strategy.ExecutingExecutionStrategy.invoke(ExecutingExecutionStrategy.java:58) > at > org.eclipse.jetty.util.thread.strategy.ExecuteProduceConsume.produceConsume(ExecuteProduceConsume.java:201) > at > org.eclipse.jetty.util.thread.strategy.ExecuteProduceConsume.run(ExecuteProduceConsume.java:133) > at > org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:672) > at > org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:590) > at java.lang.Thread.run(Thread.java:745) > Caused by: java.lang.NullPointerException > at > org.apache.jasper.servlet.JspServlet.handleMissingResource(JspServlet.java:397) > at > org.apache.jasper.servlet.JspServlet.serviceJspFile(JspServlet.java:387) > at org.apache.jasper.servlet.JspServlet.init(JspServlet.java:138) > at > org.eclipse.jetty.servlet.ServletHolder.initServlet(ServletHolder.java:637) > ... 36 more > > Caused by: > > java.lang.NullPointerException > at > org.apache.jasper.servlet.JspServlet.handleMissingResource(JspServlet.java:397) > at > org.apache.jasper.servlet.JspServlet.serviceJspFile(JspServlet.java:387) > at org.apache.jasper.servlet.JspServlet.init(JspServlet.java:138) > at > org.eclipse.jetty.servlet.ServletHolder.initServlet(ServletHolder.java:637) > at > org.eclipse.jetty.servlet.ServletHolder.getServlet(ServletHolder.java:498) > at > org.eclipse.jetty.servlet.ServletHolder.ensureInstance(ServletHolder.java:785) > at > org.eclipse.jetty.servlet.ServletHolder.prepare(ServletHolder.java:770) > at > org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:538) > at > org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:143) > at > org.eclipse.jetty.security.SecurityHandler.handle(SecurityHandler.java:548) > at > org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:132) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:190) > at > org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1593) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:188) > at > org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1239) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:168) > at > org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:481) > at > org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1562) > at > org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:166) > at > org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1141) > at > org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141) > at > org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:118) > at > org.eclipse.jetty.server.handler.gzip.GzipHandler.handle(GzipHandler.java:561) > at > org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:132) > at org.eclipse.jetty.server.Server.handle(Server.java:564) > at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:320) > at > org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:251) > at > org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:279) > at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:110) > at > org.eclipse.jetty.io.ssl.SslConnection.onFillable(SslConnection.java:258) > at > org.eclipse.jetty.io.ssl.SslConnection$3.succeeded(SslConnection.java:147) > at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:110) > at org.eclipse.jetty.io.ChannelEndPoint$2.run(ChannelEndPoint.java:124) > at > org.eclipse.jetty.util.thread.Invocable.invokePreferred(Invocable.java:122) > at > org.eclipse.jetty.util.thread.strategy.ExecutingExecutionStrategy.invoke(ExecutingExecutionStrategy.java:58) > at > org.eclipse.jetty.util.thread.strategy.ExecuteProduceConsume.produceConsume(ExecuteProduceConsume.java:201) > at > org.eclipse.jetty.util.thread.strategy.ExecuteProduceConsume.run(ExecuteProduceConsume.java:133) > at > org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:672) > at > org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:590) > at java.lang.Thread.run(Thread.java:745) > > My configuration inside nifi.properties is as below: > # Provenance Repository Properties > nifi.provenance.repository.implementation=org.apache.nifi.provenance. > WriteAheadProvenanceRepository > nifi.provenance.repository.debug.frequency=1_000_000 > nifi.provenance.repository.encryption.key.provider.implementation= > nifi.provenance.repository.encryption.key.provider.location= > nifi.provenance.repository.encryption.key.id= > nifi.provenance.repository.encryption.key= > > # Persistent Provenance Repository Properties > nifi.provenance.repository.directory.default=../provenance_repository > nifi.provenance.repository.max.storage.time=24 hours > nifi.provenance.repository.max.storage.size=1 GB > nifi.provenance.repository.rollover.time=30 secs > nifi.provenance.repository.rollover.size=100 MB > nifi.provenance.repository.query.threads=2 > nifi.provenance.repository.index.threads=1 > nifi.provenance.repository.compress.on.rollover=true > nifi.provenance.repository.always.sync=false > nifi.provenance.repository.index.shard.size=4 GB > > > By the way, does this Data Provenance list all FlowFiles ever created or > only part of it? Should I try to find the FlowFile with the exception time > in the log? Thanks. > > Regards, > Ben > > 2017-12-27 16:57 GMT+08:00 Koji Kawamura <ijokaruma...@gmail.com>: > >> Hi Ben, >> >> The ExecuteSqlCommand retry logic does not execute the same query >> multiple times if it succeeds. >> So, there must be input FlowFiles containing the same query had been >> passed more than once. >> It could be the same FlowFile, or different FlowFiles generated by the >> first processor for some reason. >> To investigate those kind of FlowFile level information, NiFi >> provenance data and FlowFile lineage will be very useful. >> https://nifi.apache.org/docs/nifi-docs/html/user-guide.html# >> viewing-flowfile-lineage >> >> I didn't mention about it earlier because you were having Provenance >> repository performance issue, but I hope you can use it now with the >> WriteAheadProvenanceRepository. >> >> Thanks, >> Koji >> >> On Wed, Dec 27, 2017 at 5:44 PM, 尹文才 <batman...@gmail.com> wrote: >> > Thanks Koji, for the ExecuteSqlCommand issue, I was trying to re-execute >> > the sql query if the connection is lost(connection could be unstable), >> my >> > idea is to only transfer the FlowFile to the success relationship >> > after successfully executing the sql query. You could see the do while >> loop >> > in the code, the transaction will be rollbacked if the execution >> failed; if >> > the connection is lost, it will retry to execute the sql. >> > Will this logic cause my sql to be executed twice? >> > >> > For the WaitBatch processor, I will take your approach to test >> individually >> > to see if the WaitBatch processor could cause the FlowFile repository >> > checkpointing failure. >> > >> > Regards, >> > Ben >> > >> > 2017-12-27 16:10 GMT+08:00 Koji Kawamura <ijokaruma...@gmail.com>: >> > >> >> Hi Ben, >> >> >> >> Excuse me, I'm trying, but probably I don't fully understand what you >> >> want to achieve with the flow. >> >> >> >> It looks weird that WaitBatch is failing with such FlowFile repository >> >> error, while other processor such as ReplaceText succeeds. >> >> I recommend to test WaitBatch alone first without combining the >> >> database related processors, by feeding a test FlowFile having >> >> expected FlowFile attributes. >> >> Such input FlowFiles can be created by GenerateFlowFile processor. >> >> If the same error happens with only WaitBatch processor, then it >> >> should be easier to debug. >> >> >> >> Thanks, >> >> Koji >> >> >> >> On Wed, Dec 27, 2017 at 4:49 PM, Koji Kawamura <ijokaruma...@gmail.com >> > >> >> wrote: >> >> > Hi Ben, >> >> > >> >> > The one thing that looks strange in the screenshot is the >> >> > ExecuteSqlCommand having FlowFiles queued in its incoming connection. >> >> > Those should be transferred to 'failure' relationship. >> >> > >> >> > Following executeSql() method, shouldn't it re-throw the caught >> >> exception? >> >> > >> >> > >> >> > try (Connection con = dbcpService.getConnection()) { >> >> > logger.debug("设置autoCommit为false"); >> >> > con.setAutoCommit(false); >> >> > >> >> > try (Statement stmt = con.createStatement()) { >> >> > logger.info("执行sql语句: {}", new Object[]{sql}); >> >> > stmt.execute(sql); >> >> > >> >> > // 所有sql语句执行在一个transaction内 >> >> > logger.debug("提交transaction"); >> >> > con.commit(); >> >> > } catch (Exception ex) { >> >> > logger.error("执行sql语句失败:{}", new Object[]{sql, >> ex}); >> >> > con.rollback(); >> >> > //将exception抛到外层处理 >> >> > throw ex; >> >> > } finally { >> >> > logger.debug("重新设置autoCommit为true"); >> >> > con.setAutoCommit(true); >> >> > } >> >> > } catch (Exception ex) { >> >> > // HERE, the exception is swallowed, that's why the FlowFiles stay in >> >> > the incoming connection. >> >> > logger.error("重试执行sql语句:{}", new Object[]{sql, ex}); >> >> > retryOnFail = true; >> >> > } >> >> > >> >> > Thanks, >> >> > Koji >> >> > >> >> > On Wed, Dec 27, 2017 at 2:38 PM, 尹文才 <batman...@gmail.com> wrote: >> >> >> Hi Koji, no problem. You could check the code of processor >> WaitBatch at >> >> the >> >> >> link: >> >> >> https://drive.google.com/open?id=1DMpW5GMiXpyZQdui989Rr3D9rlchQfWQ >> >> >> >> >> >> I also uploaded a snapshot of part of NiFi flow which includes the >> >> >> ExecuteSqlCommand and WaitBatch, you could check the picture at the >> >> link: >> >> >> https://drive.google.com/file/d/1vdxlWj8ANHQH0CMrXnydLni5o-3 >> IVi2h/view >> >> >> >> >> >> You mentioned above that FlowFile repository fails checkpointing >> will >> >> >> affect other processors to process same FlowFile again, but as you >> could >> >> >> see from my snapshot image, the ExecuteSqlCommand is the second >> >> processor >> >> >> and before the WaitBatch processor, even if the FlowFile repository >> >> >> checkpointing failure is caused by WaitBatch, could it lead to the >> >> >> processors before it to process a FlowFile multiple times? Thanks. >> >> >> >> >> >> Regards, >> >> >> Ben >> >> >> >> >> >> 2017-12-27 12:36 GMT+08:00 Koji Kawamura <ijokaruma...@gmail.com>: >> >> >> >> >> >>> Hi Ben, >> >> >>> >> >> >>> I was referring these two log messages in your previous email. >> >> >>> These two messages are both written by ExecuteSqlCommand, it does >> not >> >> >>> mean 'it was executed again'. >> >> >>> >> >> >>> ``` >> >> >>> 2017-12-26 07:00:01,312 INFO [Timer-Driven Process Thread-1] >> >> >>> c.z.nifi.processors.ExecuteSqlCommand >> >> >>> ExecuteSqlCommand[id=3c97dfd8-aaa4-3a37-626e-fed5a4822d14] >> 执行sql语句: >> >> SELECT >> >> >>> TOP 0 * INTO tmp.ods_extractDataDebug_20171226031801926_9195 FROM >> >> >>> dbo.ods_extractDataDebug; >> >> >>> alter table tmp.ods_extractDataDebug_20171226031801926_9195 drop >> >> column >> >> >>> _id; >> >> >>> >> >> >>> and it was executed again later: >> >> >>> >> >> >>> 2017-12-26 07:00:01,315 ERROR [Timer-Driven Process Thread-1] >> >> >>> c.z.nifi.processors.ExecuteSqlCommand >> >> >>> ExecuteSqlCommand[id=3c97dfd8-aaa4-3a37-626e-fed5a4822d14] >> >> >>> 执行sql语句失败:SELECT >> >> >>> ``` >> >> >>> >> >> >>> As you written, the case where FlowFile repository fails >> checkpointing >> >> >>> will affect other processors to process same FlowFiles again. >> However >> >> >>> there won't be a simple solution to every processor to rollback its >> >> >>> job as different processors do different things. Creating a temp >> table >> >> >>> if not exist seems right approach to me. >> >> >>> >> >> >>> At the same time, the route cause of getting FlowFile repository >> >> >>> failed should be investigated. Is it possible to share WaitBatch >> code? >> >> >>> The reason why ask this is all 'FlowFile Repository failed to >> update' >> >> >>> is related to WaitBatch processor in the log that you shared >> earlier. >> >> >>> >> >> >>> Thanks, >> >> >>> Koji >> >> >>> >> >> >>> On Wed, Dec 27, 2017 at 1:19 PM, 尹文才 <batman...@gmail.com> wrote: >> >> >>> > Hi Koji, I will print the sql before actually executing it, but I >> >> checked >> >> >>> > the error log line you mentioned in your reply, this error was >> >> thrown by >> >> >>> > NiFi from within another processor called WaitBatch. >> >> >>> > I didn't find similar errors as the one from the >> ExecuteSqlCommand >> >> >>> > processor, I think it's because only the ExecuteSqlCommand is >> used to >> >> >>> > create temp database tables. >> >> >>> > You could check my ExecuteSqlCommand code via the link: >> >> >>> > https://drive.google.com/open?id=1NnjBihyKpmUPEH7X28Mh2hgOrh >> jSk_5P >> >> >>> > >> >> >>> > If the error is really caused by FlowFile repository checkpoint >> >> failure >> >> >>> and >> >> >>> > the flowfile was executed twice, I may have to create the temp >> table >> >> only >> >> >>> > if doesn't exist, I didn't fix this bug in this way >> >> >>> > right away is because I was afraid this fix could cover some >> other >> >> >>> problems. >> >> >>> > >> >> >>> > Thanks. >> >> >>> > >> >> >>> > Regards, >> >> >>> > Ben >> >> >>> > >> >> >>> > 2017-12-27 11:38 GMT+08:00 Koji Kawamura <ijokaruma...@gmail.com >> >: >> >> >>> > >> >> >>> >> Hi Ben, >> >> >>> >> >> >> >>> >> The following two log messages are very close in terms of >> written >> >> >>> >> timestamp, but have different log level. >> >> >>> >> 2017-12-26 07:00:01,312 INFO >> >> >>> >> 2017-12-26 07:00:01,315 ERROR >> >> >>> >> >> >> >>> >> I guess those are logged within a single onTrigger of your >> >> >>> >> ExecuteSqlCommand custom processor, one is before executing, the >> >> other >> >> >>> >> is when it caught an exception. Just guessing as I don't have >> access >> >> >>> >> to the code. >> >> >>> >> >> >> >>> >> Does the same issue happen with other processors bundled with >> Apache >> >> >>> >> NiFi without your custom processor running? >> >> >>> >> >> >> >>> >> If NiFi fails to update/checkpoint FlowFile repository, then the >> >> same >> >> >>> >> FlowFile can be processed again after restarting NiFi. >> >> >>> >> >> >> >>> >> Thanks, >> >> >>> >> Koji >> >> >>> >> >> >> >>> >> >> >> >>> >> >> >> >>> >> On Wed, Dec 27, 2017 at 12:21 PM, 尹文才 <batman...@gmail.com> >> wrote: >> >> >>> >> > Thanks Koji, I will look into this article about the record >> model. >> >> >>> >> > >> >> >>> >> > By the way, that error I previously mentioned to you occurred >> >> again, I >> >> >>> >> > could see the sql query was executed twice in the log, this >> time >> >> I had >> >> >>> >> > turned on the verbose NiFi logging, the sql query is as below: >> >> >>> >> > >> >> >>> >> > 2017-12-26 07:00:01,312 INFO [Timer-Driven Process Thread-1] >> >> >>> >> > c.z.nifi.processors.ExecuteSqlCommand >> >> >>> >> > ExecuteSqlCommand[id=3c97dfd8-aaa4-3a37-626e-fed5a4822d14] >> >> 执行sql语句: >> >> >>> >> SELECT >> >> >>> >> > TOP 0 * INTO tmp.ods_extractDataDebug_20171226031801926_9195 >> FROM >> >> >>> >> > dbo.ods_extractDataDebug; >> >> >>> >> > alter table tmp.ods_extractDataDebug_20171226031801926_9195 >> drop >> >> >>> column >> >> >>> >> _id; >> >> >>> >> > >> >> >>> >> > and it was executed again later: >> >> >>> >> > >> >> >>> >> > 2017-12-26 07:00:01,315 ERROR [Timer-Driven Process Thread-1] >> >> >>> >> > c.z.nifi.processors.ExecuteSqlCommand >> >> >>> >> > ExecuteSqlCommand[id=3c97dfd8-aaa4-3a37-626e-fed5a4822d14] >> >> >>> >> 执行sql语句失败:SELECT >> >> >>> >> > TOP 0 * INTO tmp.ods_extractDataDebug_20171226031801926_9195 >> FROM >> >> >>> >> > dbo.ods_extractDataDebug; >> >> >>> >> > alter table tmp.ods_extractDataDebug_20171226031801926_9195 >> drop >> >> >>> column >> >> >>> >> > _id;: com.microsoft.sqlserver.jdbc.SQLServerException: >> 数据库中已存在名为 >> >> >>> >> > 'ods_extractDataDebug_20171226031801926_9195' 的对象。 >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerException: 数据库中已存在名为 >> >> >>> >> > 'ods_extractDataDebug_20171226031801926_9195' 的对象。 >> >> >>> >> > at >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerException. >> >> >>> makeFromDatabaseError( >> >> >>> >> SQLServerException.java:217) >> >> >>> >> > at >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerStatement.getNextResul >> t( >> >> >>> >> SQLServerStatement.java:1655) >> >> >>> >> > at >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerStatement. >> >> doExecuteStatement( >> >> >>> >> SQLServerStatement.java:885) >> >> >>> >> > at >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerStatement$ >> >> >>> StmtExecCmd.doExecute( >> >> >>> >> SQLServerStatement.java:778) >> >> >>> >> > at com.microsoft.sqlserver.jdbc.TDSCommand.execute(IOBuffer. >> >> >>> java:7505) >> >> >>> >> > at >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerConnection.executeComm >> and( >> >> >>> >> SQLServerConnection.java:2445) >> >> >>> >> > at >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerStatement.executeComma >> nd( >> >> >>> >> SQLServerStatement.java:191) >> >> >>> >> > at >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerStatement.executeState >> ment( >> >> >>> >> SQLServerStatement.java:166) >> >> >>> >> > at >> >> >>> >> > com.microsoft.sqlserver.jdbc.SQLServerStatement.execute( >> >> >>> >> SQLServerStatement.java:751) >> >> >>> >> > at >> >> >>> >> > org.apache.commons.dbcp.DelegatingStatement.execute( >> >> >>> >> DelegatingStatement.java:264) >> >> >>> >> > at >> >> >>> >> > org.apache.commons.dbcp.DelegatingStatement.execute( >> >> >>> >> DelegatingStatement.java:264) >> >> >>> >> > at >> >> >>> >> > com.zjrealtech.nifi.processors.ExecuteSqlCommand. >> >> >>> >> executeSql(ExecuteSqlCommand.java:194) >> >> >>> >> > at >> >> >>> >> > com.zjrealtech.nifi.processors.ExecuteSqlCommand. >> >> >>> >> onTrigger(ExecuteSqlCommand.java:164) >> >> >>> >> > at >> >> >>> >> > org.apache.nifi.processor.AbstractProcessor.onTrigger( >> >> >>> >> AbstractProcessor.java:27) >> >> >>> >> > at >> >> >>> >> > org.apache.nifi.controller.StandardProcessorNode.onTrigger( >> >> >>> >> StandardProcessorNode.java:1119) >> >> >>> >> > at >> >> >>> >> > org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask. >> >> call( >> >> >>> >> ContinuallyRunProcessorTask.java:147) >> >> >>> >> > at >> >> >>> >> > org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask. >> >> call( >> >> >>> >> ContinuallyRunProcessorTask.java:47) >> >> >>> >> > at >> >> >>> >> > org.apache.nifi.controller.scheduling. >> >> TimerDrivenSchedulingAgent$1. >> >> >>> run( >> >> >>> >> TimerDrivenSchedulingAgent.java:128) >> >> >>> >> > at java.util.concurrent.Executors$RunnableAdapter. >> >> >>> >> call(Executors.java:511) >> >> >>> >> > at java.util.concurrent.FutureTask.runAndReset( >> >> FutureTask.java:308) >> >> >>> >> > at >> >> >>> >> > java.util.concurrent.ScheduledThreadPoolExecutor$ >> >> >>> >> ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor. >> >> java:180) >> >> >>> >> > at >> >> >>> >> > java.util.concurrent.ScheduledThreadPoolExecutor$ >> >> >>> >> ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) >> >> >>> >> > 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) >> >> >>> >> > >> >> >>> >> > I also saw a lot of NiFi's exception like "ProcessException: >> >> FlowFile >> >> >>> >> > Repository failed to update", not sure if this is the reason >> the >> >> >>> FlowFile >> >> >>> >> > got processed twice. Could you help to take a look at my log >> >> file? >> >> >>> >> Thanks. >> >> >>> >> > You could get the log file via the link: >> >> >>> >> > https://drive.google.com/file/d/1uVgtAVNEHxAbAPEpNTOWq_ >> >> >>> N9Xu6zMEi3/view >> >> >>> >> > >> >> >>> >> > Best Regards, >> >> >>> >> > Ben >> >> >>> >> > >> >> >>> >> > 2017-12-27 10:00 GMT+08:00 Koji Kawamura < >> ijokaruma...@gmail.com >> >> >: >> >> >>> >> > >> >> >>> >> >> Hi Ben, >> >> >>> >> >> >> >> >>> >> >> This blog post written by Mark, would be a good starting >> point >> >> to get >> >> >>> >> >> familiar with NiFi Record model. >> >> >>> >> >> https://blogs.apache.org/nifi/entry/record-oriented-data- >> >> with-nifi >> >> >>> >> >> >> >> >>> >> >> HA for DistributedMapCacheClientService and >> >> >>> DistributedMapCacheServer >> >> >>> >> >> pair is not supported at the moment. If you need >> >> HighAvailability, >> >> >>> >> >> RedisDistributedMapCacheClientService with Redis replication >> >> will >> >> >>> >> >> provide that, I haven't tried that myself though. >> >> >>> >> >> https://redis.io/topics/replication >> >> >>> >> >> >> >> >>> >> >> Thanks, >> >> >>> >> >> Koji >> >> >>> >> >> >> >> >>> >> >> On Tue, Dec 26, 2017 at 7:58 PM, 尹文才 <batman...@gmail.com> >> >> wrote: >> >> >>> >> >> > Thanks for your quick response, Koji, I haven't heard and >> seen >> >> >>> >> anything >> >> >>> >> >> > about the NiFi record data model when I was reading the >> NiFi >> >> >>> >> >> > documentations,could you tell me where this model is >> >> documented? >> >> >>> >> Thanks. >> >> >>> >> >> > >> >> >>> >> >> > By the way, to my knowledge, when you need to use the >> >> >>> >> >> DistributedMapCacheServer >> >> >>> >> >> > from DistributedMapCacheClientService, you need to >> specify the >> >> >>> host >> >> >>> >> url >> >> >>> >> >> for >> >> >>> >> >> > the server, this means inside a NiFi cluster >> >> >>> >> >> > when I specify the cache server and the node suddenly went >> >> down, I >> >> >>> >> >> couldn't >> >> >>> >> >> > possibly use it until the node goes up again right? Is >> there >> >> >>> currently >> >> >>> >> >> such >> >> >>> >> >> > a cache server in NiFi that could support HA? Thanks. >> >> >>> >> >> > >> >> >>> >> >> > Regards, >> >> >>> >> >> > Ben >> >> >>> >> >> > >> >> >>> >> >> > 2017-12-26 18:34 GMT+08:00 Koji Kawamura < >> >> ijokaruma...@gmail.com>: >> >> >>> >> >> > >> >> >>> >> >> >> Hi Ben, >> >> >>> >> >> >> >> >> >>> >> >> >> As you found from existing code, DistributedMapCache is >> used >> >> to >> >> >>> share >> >> >>> >> >> >> state among different processors, and it can be used by >> your >> >> >>> custom >> >> >>> >> >> >> processors, too. >> >> >>> >> >> >> However, I'd recommend to avoid such tight dependencies >> >> between >> >> >>> >> >> >> FlowFiles if possible, or minimize the part in flow that >> >> requires >> >> >>> >> that >> >> >>> >> >> >> constraint at least for better performance and simplicity. >> >> >>> >> >> >> For example, since a FlowFile can hold fairly large >> amount of >> >> >>> data, >> >> >>> >> >> >> you could merge all FlowFiles in a single FlowFile, >> instead of >> >> >>> >> batches >> >> >>> >> >> >> of FlowFiles. If you need logical boundaries, you can use >> NiFi >> >> >>> Record >> >> >>> >> >> >> data model to embed multiple records within a FlowFile, >> Record >> >> >>> should >> >> >>> >> >> >> perform better. >> >> >>> >> >> >> >> >> >>> >> >> >> Hope this helps. >> >> >>> >> >> >> >> >> >>> >> >> >> Thanks, >> >> >>> >> >> >> Koji >> >> >>> >> >> >> >> >> >>> >> >> >> >> >> >>> >> >> >> On Tue, Dec 26, 2017 at 5:55 PM, 尹文才 <batman...@gmail.com >> > >> >> wrote: >> >> >>> >> >> >> > Hi guys, I'm currently trying to find a proper way in >> nifi >> >> which >> >> >>> >> could >> >> >>> >> >> >> sync >> >> >>> >> >> >> > status between my custom processors. >> >> >>> >> >> >> > our requirement is like this, we're doing some ETL work >> >> using >> >> >>> nifi >> >> >>> >> and >> >> >>> >> >> >> I'm >> >> >>> >> >> >> > extracting the data from DB into batches of >> FlowFiles(each >> >> >>> batch of >> >> >>> >> >> >> > FlowFile has a flag FlowFile indicating the end of the >> >> batch). >> >> >>> >> >> >> > There're some groups of custom processors downstream >> that >> >> need >> >> >>> to >> >> >>> >> >> process >> >> >>> >> >> >> > these FlowFiles to do some business logic work. And we >> >> expect >> >> >>> these >> >> >>> >> >> >> > processors to process one batch of FlowFiles at a time. >> >> >>> >> >> >> > Therefore we need to implement a custom Wait >> processor(let's >> >> >>> just >> >> >>> >> >> call it >> >> >>> >> >> >> > WaitBatch here) to hold all the other batches of >> FlowFiles >> >> while >> >> >>> >> the >> >> >>> >> >> >> > business processors were handling the batch of FlowFiles >> >> whose >> >> >>> >> >> creation >> >> >>> >> >> >> > time is earlier. >> >> >>> >> >> >> > >> >> >>> >> >> >> > In order to implement this, all the WaitBatch processors >> >> placed >> >> >>> in >> >> >>> >> the >> >> >>> >> >> >> flow >> >> >>> >> >> >> > need to read/update records in a shared map so that each >> >> set of >> >> >>> >> >> >> > business-logic processors process one batch at a time. >> >> >>> >> >> >> > The entries are keyed using the batch number of the >> >> FlowFiles >> >> >>> and >> >> >>> >> the >> >> >>> >> >> >> value >> >> >>> >> >> >> > of each entry is a batch release counter number which >> >> counts the >> >> >>> >> >> number >> >> >>> >> >> >> of >> >> >>> >> >> >> > times the batch of FlowFiles has passed through >> >> >>> >> >> >> > a WaitBatch processor. >> >> >>> >> >> >> > When a batch is released by WaitBatch, it will try to >> >> increment >> >> >>> the >> >> >>> >> >> batch >> >> >>> >> >> >> > number entry's value by 1 and then the released batch >> >> number and >> >> >>> >> >> counter >> >> >>> >> >> >> > number will also be saved locally at the WaitBatch with >> >> >>> >> StateManager; >> >> >>> >> >> >> > when the next batch reaches the WaitBatch, it will >> check if >> >> the >> >> >>> >> >> counter >> >> >>> >> >> >> > value of the previous released batch number in the >> shared >> >> map is >> >> >>> >> >> greater >> >> >>> >> >> >> > than the one saved locally, if the entry for the batch >> >> number >> >> >>> >> does't >> >> >>> >> >> >> > exist(already removed) or the value in the shared map is >> >> >>> greater, >> >> >>> >> the >> >> >>> >> >> >> next >> >> >>> >> >> >> > batch will be released and the local state and the >> entry on >> >> the >> >> >>> >> shared >> >> >>> >> >> >> map >> >> >>> >> >> >> > will be updated similarly. >> >> >>> >> >> >> > In the end of the flow, a custom processor will get the >> >> batch >> >> >>> >> number >> >> >>> >> >> from >> >> >>> >> >> >> > each batch and remove the entry from the shared map . >> >> >>> >> >> >> > >> >> >>> >> >> >> > So this implementation requires a shared map that could >> >> >>> read/update >> >> >>> >> >> >> > frequently and atomically. I checked the Wait/Notify >> >> processors >> >> >>> in >> >> >>> >> >> NIFI >> >> >>> >> >> >> and >> >> >>> >> >> >> > saw it is using the DistributedMapCacheClientService >> and >> >> >>> >> >> >> > DistributedMapCacheServer to sync status, so I'm >> wondering >> >> if I >> >> >>> >> could >> >> >>> >> >> use >> >> >>> >> >> >> > the DistributedMapCacheClientService to implement my >> >> logic. I >> >> >>> also >> >> >>> >> >> saw >> >> >>> >> >> >> > another implementation called >> RedisDistributedMapCacheClient >> >> >>> >> Service >> >> >>> >> >> >> > which seems to require Redis(I haven't used Redis). >> Thanks >> >> in >> >> >>> >> advance >> >> >>> >> >> >> for >> >> >>> >> >> >> > any suggestions. >> >> >>> >> >> >> > >> >> >>> >> >> >> > Regards, >> >> >>> >> >> >> > Ben >> >> >>> >> >> >> >> >> >>> >> >> >> >> >>> >> >> >> >>> >> >> >> > >
