Ohara: Show error message when start/stop connector

[eechih]

Describe the bug
啟動 Ftp source connector 後，右上角會出現紅色的錯誤提示訊息：java.io.NotSerializableException:com.island.ohara.metrics.kafka.TopicMeter

Operation System (please complete the following information):

• Backend OS & Version: CentOS 7.2
• Frontend OS & Version: macOS 10.14.1
• Browser & Version: Chrome 74.0.3729.169

Ohara Version

• 0.5-SNAPSHOT
• Build on 2019-06-03 06:48:00

see #1413 for the root cause. (updated by @chia7712)

[eechih]

ping @konekoya @wu87988622 麻煩幫忙看一下這個問題，謝謝。

[wu87988622]

ping @konekoya @wu87988622 麻煩幫忙看一下這個問題，謝謝。

got it

[eechih]

補充：每次修改 FtpSourceConnector 任何欄位的內容，也都會出現相同錯誤。

[konekoya]

@wu87988622 Could you help tackle this issue? This is a critical bug and should be fixed in both 0.6.0 and 0.5.0

[eechih]

ping @chia7712 Please take a look, thanks.

Jun 05, 2019 8:54:30 AM com.island.ohara.configurator.Configurator$$anonfun$exceptionHandler$1 $anonfun$applyOrElse$1
SEVERE: Request to http://10.2.0.24:12345/v0/pipelines/3886b930-7bd9-4d8e-90b1-9c2a640c2b5f with HttpEntity.Strict(application/json,{"id":"3886b930-7bd9-4d8e-90b1-9c2a640c2b5f","name":"Untitled pipeline","status":"Stopped","workerClusterName":"k8soccl","rules":{"dfb9db54-f389-4cc8-a5b3-6bed68f43343":["T"],"T":[]}}) is wrong
java.lang.IllegalArgumentException: java.io.NotSerializableException: com.island.ohara.metrics.kafka.TopicMeter
    at com.island.ohara.common.data.Serializer$11.to(Serializer.java:405)
    at com.island.ohara.configurator.store.DataStore$Builder$$anon$1.to(DataStore.scala:149)
    at com.island.ohara.configurator.store.DataStore$Builder$$anon$1.to(DataStore.scala:148)
    at com.island.ohara.configurator.store.RocksDataStore.toValue(RocksDataStore.scala:93)
    at com.island.ohara.configurator.store.RocksDataStore.$anonfun$addIfPresent$1(RocksDataStore.scala:125)
    at scala.util.Success.$anonfun$map$1(Try.scala:255)
    at scala.util.Success.map(Try.scala:213)
    at scala.concurrent.Future.$anonfun$map$1(Future.scala:292)
    at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)
    at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
    at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.io.NotSerializableException: com.island.ohara.metrics.kafka.TopicMeter
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at scala.collection.immutable.List$SerializationProxy.writeObject(List.scala:477)
    at sun.reflect.GeneratedMethodAccessor43.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1140)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at com.island.ohara.common.data.Serializer$11.to(Serializer.java:401)
    ... 15 more

[chia7712]

@eechih Could you share more error stack with me? BTW, could you reproduce this bug in 0.6-SNAPSHOT?

[eechih]

@chia7712 I also tested the 0.6-SNAPSHOT version, the same result:

• Fill in the following data into the FtpSourceConnector form.

input folder: /demo_folder/input
complete folder: /demo_folder/complete
error folder: /demo_folder/error
hostname: 10.2.0.28
port: 21
user" ***
password: ***
Topics: T

• Click the FtpSourceConnector graph in pipeline. Configurator throw the following error:

Jun 06, 2019 1:53:59 AM com.island.ohara.configurator.route.ConnectorRoute$$anonfun$update$2 applyOrElse
SEVERE: failed to fetch stats for ConnectorDescription(3bc5c740-b288-457d-bf8e-453ea252a0e6,Map(name -> "Untitled source", workerClusterName -> "wk00", tasks.max -> 1, connector.class -> "com.island.ohara.connector.ftp.FtpSource", connector.name -> "Untitled source"),None,None,Metrics(List()),1559786036592)
java.lang.IllegalArgumentException: No status found for connector 3bc5c740-b288-457d-bf8e-453ea252a0e6
    at com.island.ohara.client.HttpExecutor$HttpExecutorImpl.$anonfun$asError$1(HttpExecutor.scala:152)
    at scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:307)
    at akka.http.scaladsl.util.FastFuture$.$anonfun$transformWith$1(FastFuture.scala:37)
    at akka.http.scaladsl.util.FastFuture$.strictTransform$1(FastFuture.scala:41)
    at akka.http.scaladsl.util.FastFuture$.transformWith$extension1(FastFuture.scala:45)
    at akka.http.scaladsl.util.FastFuture$.transformWith$extension0(FastFuture.scala:37)
    at akka.http.scaladsl.util.FastFuture$FulfilledFuture.transformWith(FastFuture.scala:86)
    at scala.concurrent.Future.flatMap(Future.scala:306)
    at scala.concurrent.Future.flatMap$(Future.scala:306)
    at akka.http.scaladsl.util.FastFuture$FulfilledFuture.flatMap(FastFuture.scala:77)
    at com.island.ohara.client.HttpExecutor$HttpExecutorImpl.asError(HttpExecutor.scala:148)
    at com.island.ohara.client.HttpExecutor$HttpExecutorImpl.unmarshal(HttpExecutor.scala:138)
    at com.island.ohara.client.HttpExecutor$HttpExecutorImpl.$anonfun$get$1(HttpExecutor.scala:167)
    at scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:307)
    at scala.concurrent.impl.Promise.$anonfun$transformWith$1(Promise.scala:41)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
    at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

• First, press the button to start the FtpSourceConnector. Configurator throw the following error:

Jun 06, 2019 1:55:58 AM com.island.ohara.client.kafka.WorkerClient$$anon$2$$anonfun$com$island$ohara$client$kafka$WorkerClient$$nestedInanon$$retry$1 applyOrElse
INFO: fetch active connectors 0/3
java.net.HttpRetryException: Cannot complete request momentarily due to stale configuration (typically caused by a concurrent config change)
    at com.island.ohara.client.HttpExecutor$HttpExecutorImpl.$anonfun$asError$1(HttpExecutor.scala:151)
    at scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:307)
    at akka.http.scaladsl.util.FastFuture$.$anonfun$transformWith$1(FastFuture.scala:37)
    at akka.http.scaladsl.util.FastFuture$.strictTransform$1(FastFuture.scala:41)
    at akka.http.scaladsl.util.FastFuture$.transformWith$extension1(FastFuture.scala:45)
    at akka.http.scaladsl.util.FastFuture$.transformWith$extension0(FastFuture.scala:37)
    at akka.http.scaladsl.util.FastFuture$FulfilledFuture.transformWith(FastFuture.scala:86)
    at scala.concurrent.Future.flatMap(Future.scala:306)
    at scala.concurrent.Future.flatMap$(Future.scala:306)
    at akka.http.scaladsl.util.FastFuture$FulfilledFuture.flatMap(FastFuture.scala:77)
    at com.island.ohara.client.HttpExecutor$HttpExecutorImpl.asError(HttpExecutor.scala:148)
    at com.island.ohara.client.HttpExecutor$HttpExecutorImpl.unmarshal(HttpExecutor.scala:138)
    at com.island.ohara.client.HttpExecutor$HttpExecutorImpl.$anonfun$get$1(HttpExecutor.scala:167)
    at scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:307)
    at scala.concurrent.impl.Promise.$anonfun$transformWith$1(Promise.scala:41)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
    at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

• Press the button to start the FtpSourceConnector again and the connector starts up normally.

• Press the button to stop the FtpSourceConnector and the connector stops normally.

• Put a CSV file into the FTP input folder, and the connector successfully feed the data to topic.

• Press the button to stop the FtpSourceConnector, and modify any of the fields. Configurator throw the following error:

Jun 06, 2019 2:09:24 AM com.island.ohara.configurator.Configurator$$anonfun$exceptionHandler$1 $anonfun$applyOrElse$1
SEVERE: Request to http://10.2.0.24:12345/v0/pipelines/4aeab2db-8912-4a85-8c3b-cf8569cbdf72 with HttpEntity.Strict(application/json,{"id":"4aeab2db-8912-4a85-8c3b-cf8569cbdf72","name":"Untitled pipeline","status":"Stopped","workerClusterName":"wk00","rules":{"3bc5c740-b288-457d-bf8e-453ea252a0e6":["T"],"T":[]}}) is wrong
java.lang.IllegalArgumentException: java.io.NotSerializableException: com.island.ohara.metrics.kafka.TopicMeter
    at com.island.ohara.common.data.Serializer$11.to(Serializer.java:405)
    at com.island.ohara.configurator.store.DataStore$Builder$$anon$1.to(DataStore.scala:149)
    at com.island.ohara.configurator.store.DataStore$Builder$$anon$1.to(DataStore.scala:148)
    at com.island.ohara.configurator.store.RocksDataStore.toValue(RocksDataStore.scala:93)
    at com.island.ohara.configurator.store.RocksDataStore.$anonfun$addIfPresent$1(RocksDataStore.scala:125)
    at scala.util.Success.$anonfun$map$1(Try.scala:255)
    at scala.util.Success.map(Try.scala:213)
    at scala.concurrent.Future.$anonfun$map$1(Future.scala:292)
    at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)
    at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
    at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.io.NotSerializableException: com.island.ohara.metrics.kafka.TopicMeter
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at scala.collection.immutable.List$SerializationProxy.writeObject(List.scala:477)
    at sun.reflect.GeneratedMethodAccessor39.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1140)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at com.island.ohara.common.data.Serializer$11.to(Serializer.java:401)
    ... 15 more

[eechih]

@chia7712 一但出現 java.lang.IllegalArgumentException: java.io.NotSerializableException: com.island.ohara.metrics.kafka.TopicMeter 後，就無法再新增任何的 topic 到 pipeline。

Request URL: http://demo00:5050/api/pipelines/603413a9-8697-4d28-a2d7-1565320ed40b
Request Method: PUT
Status Code: 400 Bad Request
Remote Address: 10.2.0.24:5050
Referrer Policy: no-referrer-when-downgrade
Request Payload:
{"id":"603413a9-8697-4d28-a2d7-1565320ed40b","name":"Test2","status":"Stopped","workerClusterName":"wk00","rules":{"47298fa9-86f9-4721-b616-289864f19632":[],"T":[]}}

Configurator server: 10.2.0.24
Broker cluster: 10.2.0.25
Worker cluster: 10.2.0.25

[chia7712]

@eechih see #1413 for the root cause.

[chia7712]

@eechih Please confirm the fix on 0.5 and then close this issue if all work well.

[eechih]

@chia7712 Thanks.