[
https://issues.apache.org/jira/browse/SAMZA-1371?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jagadish updated SAMZA-1371:
----------------------------
Description:
We have multiple Samza apps using local store that have this issue. Some
containers get stuck on "Starting BrokerProxy for hostname:portnum" while
others seem to work as expected.
Here is the log:
stuck:
```
[...]
2017-07-25 17:11:26.546 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Creating new SimpleConsumer for host hostname:portnum for system kafka
2017-07-25 17:11:26.547 [main] org.apache.samza.system.kafka.GetOffset [INFO]
Validating offset 0 for topic and partition
[prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_matcherValidation,2]
2017-07-25 17:11:26.648 [main] org.apache.samza.system.kafka.GetOffset [INFO]
Able to successfully read from offset 0 for topic and partition
[prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_matcherValidation,2].
Using it to instantiate consumer.
2017-07-25 17:11:26.649 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Starting BrokerProxy for hostname:portnum
// it's dead, Jim
```
healthy:
```
[...]
2017-07-25 17:11:26.920 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Creating new SimpleConsumer for host hostname:portnum for system kafka
2017-07-25 17:11:26.921 [main] org.apache.samza.system.kafka.GetOffset [INFO]
Validating offset 0 for topic and partition
[prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_matcherValidation,1]
2017-07-25 17:11:27.023 [main] org.apache.samza.system.kafka.GetOffset [INFO]
Able to successfully read from offset 0 for topic and partition
[prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_matcherValidation,1].
Using it to instantiate consumer.
2017-07-25 17:11:27.023 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Starting BrokerProxy for hostname:portnum
2017-07-25 17:11:29.194 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Shutting down BrokerProxy for hostname:portnum
2017-07-25 17:11:29.194 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
closing simple consumer...
2017-07-25 17:11:29.239 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.DefaultFetchSimpleConsumer [INFO] Reconnect due
to socket error: java.nio.channels.ClosedChannelException
2017-07-25 17:11:29.244 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.BrokerProxy [WARN] Restarting consumer due to
java.nio.channels.ClosedChannelException. Releasing ownership of all
partitions, and restarting consumer. Turn on debugging to get a full stack
trace.
2017-07-25 17:11:29.247 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.KafkaSystemConsumer [INFO] Abdicating for
[prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_alertSetting,1]
2017-07-25 17:11:29.247 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.KafkaSystemConsumer [INFO] Refreshing brokers
for:
Map([prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_alertSetting,1]
-> 13572)
2017-07-25 17:11:29.247 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to interrupt.
2017-07-25 17:11:29.247 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Shutting down BrokerProxy for hostname:portnum
2017-07-25 17:11:29.248 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
closing simple consumer...
2017-07-25 17:11:29.265 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to interrupt.
2017-07-25 17:11:29.265 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Shutting down BrokerProxy for hostname:portnum
2017-07-25 17:11:29.265 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
closing simple consumer...
2017-07-25 17:11:29.523 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to interrupt.
2017-07-25 17:11:29.524 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Shutting down BrokerProxy for hostname:portnum
2017-07-25 17:11:29.524 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
closing simple consumer...
2017-07-25 17:11:29.601 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to interrupt.
2017-07-25 17:11:29.602 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
Shutting down BrokerProxy for hostname:portnum
2017-07-25 17:11:29.602 [main] org.apache.samza.system.kafka.BrokerProxy [INFO]
closing simple consumer...
2017-07-25 17:11:29.663 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
hostname:portnum for client
samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to interrupt.
2017-07-25 17:11:29.668 [main] org.apache.samza.container.SamzaContainer [INFO]
Starting host statistics monitor
2017-07-25 17:11:29.670 [main] org.apache.samza.container.SamzaContainer [INFO]
Registering task instances with producers.
2017-07-25 17:11:29.674 [main] org.apache.samza.container.SamzaContainer [INFO]
Starting producer multiplexer.
2017-07-25 17:11:29.675 [main] org.apache.samza.container.SamzaContainer [INFO]
Initializing stream tasks.
2017-07-25 17:11:29.676 [main] com.company.samza.app.companyStreamingAppWrapper
[INFO] Initializing instance of streaming application
2017-07-25 17:11:29.681 [main] com.company.samza.app.companyStreamingAppWrapper
[INFO] First initialization. Setting up Guice container with configuration
companyStreamingAppWrapperConfiguration{company.app.name=ListingAlertsEmailOrganizerInstant,
company.appgroup=aws, company.env=prod,
company.guice.module=com.company.notifications.listingAlerts.organizer.email.ListingAlertsEmailOrganizerModule}
2017-07-25 17:11:30.118 [main] com.company.config.guice.configModule [INFO]
configModule loaded requested override file
'/storage/data/secure/config/AnalyticsServiceClient.cfg'
2017-07-25 17:11:30.480 [main]
com.company.samza.dataService.SamzaSessionFactoriesModule [INFO] Loading prod
dbConfig from /data/config/prod.database.properties
// Hibernate stuff (i.e. our code is hit)
```
was:
I noticed this behavior in the AM today where it fails to allocate a new
container on failure to start. The AM seems to get a callback for a container
failure during container startup and is never rescheduled again by the AM. The
logs seem to suggest that the request was made by the AM to start container
`container_e20_1528615592911_1987_02_000062` as seen below:
{code}
2018-08-06 15:39:46.369 [Container Allocator Thread] YarnClusterResourceManager
[INFO] Received launch request for 12 on hostname lca1-app0596.stg.linkedin.com
2018-08-06 15:39:46.974 [Container Allocator Thread] YarnClusterResourceManager
[INFO] Got available container ID (12) for container: Container: [ContainerId:
container_e20_1528615592911_1987_02_000062, NodeId:
lca1-app0596.stg.linkedin.com:1158, NodeHttpAddress:
lca1-app0596.stg.linkedin.com:8042, Resource: <memory:4096, vCores:1>,
Priority: 1, Token: Token { kind: ContainerToken, service: 10.251.166.210:1158
}, ]
2018-08-06 15:39:46.974 [Container Allocator Thread] YarnClusterResourceManager
[INFO] In runContainer in util: fwkPath= ;cmdPath=./__package/;jobLib=
2018-08-06 15:39:46.974 [Container Allocator Thread] YarnClusterResourceManager
[INFO] Container ID 12 using command ./__package//bin/run-container.sh
2018-08-06 15:39:46.975 [Container Allocator Thread] YarnClusterResourceManager
[INFO] Container ID 12 using environment variables:
SAMZA_CONTAINER_ID=12
EXECUTION_ENV_CONTAINER_ID=container_e20_1528615592911_1987_02_000062
SAMZA_COORDINATOR_URL=http://lca1-app1576.stg.linkedin.com:44222/
JAVA_OPTS=-Xmx3072m -Dcom.linkedin.app.name=ad-web-analytics-event-aggregator
-Dcom.linkedin.app.env=ei-lca1
{code}
But due to the issue where the localization phase fails (currently under
investigation) the container fails to start, throwing the following exception
to the `onStartContainerError` callback in `YarnClusterResourceManager`.
{code}
2018-08-06 15:46:15.257
[org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl #6]
YarnClusterResourceManager [ERROR] Container:
container_e20_1528615592911_1987_02_000062 could not start.
java.net.ConnectException: Call From
lca1-app1576.stg.linkedin.com/10.251.174.104 to
lca1-app0596.stg.linkedin.com:1158 failed on connection exception:
java.net.ConnectException: Connection refused; For more details see:
http://wiki.apache.org/hadoop/ConnectionRefused
at sun.reflect.GeneratedConstructorAccessor374.newInstance(Unknown Source)
at
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at org.apache.hadoop.net.NetUtils.wrapWithMessage(NetUtils.java:791)
at org.apache.hadoop.net.NetUtils.wrapException(NetUtils.java:731)
at org.apache.hadoop.ipc.Client.call(Client.java:1473)
at org.apache.hadoop.ipc.Client.call(Client.java:1400)
at
org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:232)
at com.sun.proxy.$Proxy34.startContainers(Unknown Source)
at
org.apache.hadoop.yarn.api.impl.pb.client.ContainerManagementProtocolPBClientImpl.startContainers(ContainerManagementProtocolPBClientImpl.java:96)
at sun.reflect.GeneratedMethodAccessor19.invoke(Unknown Source)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187)
at
org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
at com.sun.proxy.$Proxy35.startContainers(Unknown Source)
at
org.apache.hadoop.yarn.client.api.impl.NMClientImpl.startContainer(NMClientImpl.java:201)
at
org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl$StatefulContainer$StartContainerTransition.transition(NMClientAsyncImpl.java:377)
at
org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl$StatefulContainer$StartContainerTransition.transition(NMClientAsyncImpl.java:363)
at
org.apache.hadoop.yarn.state.StateMachineFactory$MultipleInternalArc.doTransition(StateMachineFactory.java:385)
at
org.apache.hadoop.yarn.state.StateMachineFactory.doTransition(StateMachineFactory.java:302)
at
org.apache.hadoop.yarn.state.StateMachineFactory.access$300(StateMachineFactory.java:46)
at
org.apache.hadoop.yarn.state.StateMachineFactory$InternalStateMachine.doTransition(StateMachineFactory.java:448)
at
org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl$StatefulContainer.handle(NMClientAsyncImpl.java:498)
at
org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl$ContainerEventProcessor.run(NMClientAsyncImpl.java:557)
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)
Caused by: java.net.ConnectException: Connection refused
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
at
org.apache.hadoop.net.SocketIOWithTimeout.connect(SocketIOWithTimeout.java:206)
at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:530)
at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:494)
at org.apache.hadoop.ipc.Client$Connection.setupConnection(Client.java:608)
at org.apache.hadoop.ipc.Client$Connection.setupIOstreams(Client.java:706)
at org.apache.hadoop.ipc.Client$Connection.access$2800(Client.java:369)
at org.apache.hadoop.ipc.Client.getConnection(Client.java:1522)
at org.apache.hadoop.ipc.Client.call(Client.java:1439)
... 22 more
2018-08-06 15:46:15.259
[org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl #6]
YarnClusterResourceManager [INFO] Got an invalid notification for container:
container_e20_1528615592911_1987_02_000062
{code}
Looking at the code in `onStartContainerError`: it seems like we are picking a
container from a Map called ` containersPendingStartup` which doesn’t seem to
have valid containers (quick code search did not show when this map is
populated). This causes these failure callbacks to go unaddressed leaving the
job to have containers that are always pending and the only resolution is to
bounce the job.
> Some Samza Containers get stuck at "Starting BrokerProxy for
> hostname:portnum" while others seem to be fine
> -----------------------------------------------------------------------------------------------------------
>
> Key: SAMZA-1371
> URL: https://issues.apache.org/jira/browse/SAMZA-1371
> Project: Samza
> Issue Type: Bug
> Components: container
> Affects Versions: 0.11.0, 0.12.0
> Environment: Samza version: 0.11, 0.12
> Kafka version: 0.11.0.0
> Reporter: Ak Ka
> Priority: Blocker
> Attachments: stdout.log, thread_dump.txt
>
>
> We have multiple Samza apps using local store that have this issue. Some
> containers get stuck on "Starting BrokerProxy for hostname:portnum" while
> others seem to work as expected.
> Here is the log:
> stuck:
> ```
> [...]
> 2017-07-25 17:11:26.546 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Creating new SimpleConsumer for host hostname:portnum for system kafka
> 2017-07-25 17:11:26.547 [main] org.apache.samza.system.kafka.GetOffset [INFO]
> Validating offset 0 for topic and partition
> [prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_matcherValidation,2]
> 2017-07-25 17:11:26.648 [main] org.apache.samza.system.kafka.GetOffset [INFO]
> Able to successfully read from offset 0 for topic and partition
> [prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_matcherValidation,2].
> Using it to instantiate consumer.
> 2017-07-25 17:11:26.649 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Starting BrokerProxy for hostname:portnum
> // it's dead, Jim
> ```
> healthy:
> ```
> [...]
> 2017-07-25 17:11:26.920 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Creating new SimpleConsumer for host hostname:portnum for system kafka
> 2017-07-25 17:11:26.921 [main] org.apache.samza.system.kafka.GetOffset [INFO]
> Validating offset 0 for topic and partition
> [prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_matcherValidation,1]
> 2017-07-25 17:11:27.023 [main] org.apache.samza.system.kafka.GetOffset [INFO]
> Able to successfully read from offset 0 for topic and partition
> [prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_matcherValidation,1].
> Using it to instantiate consumer.
> 2017-07-25 17:11:27.023 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Starting BrokerProxy for hostname:portnum
> 2017-07-25 17:11:29.194 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Shutting down BrokerProxy for hostname:portnum
> 2017-07-25 17:11:29.194 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] closing simple consumer...
> 2017-07-25 17:11:29.239 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.DefaultFetchSimpleConsumer [INFO] Reconnect due
> to socket error: java.nio.channels.ClosedChannelException
> 2017-07-25 17:11:29.244 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.BrokerProxy [WARN] Restarting consumer due to
> java.nio.channels.ClosedChannelException. Releasing ownership of all
> partitions, and restarting consumer. Turn on debugging to get a full stack
> trace.
> 2017-07-25 17:11:29.247 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.KafkaSystemConsumer [INFO] Abdicating for
> [prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_alertSetting,1]
> 2017-07-25 17:11:29.247 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.KafkaSystemConsumer [INFO] Refreshing brokers
> for:
> Map([prod.localStateChangeLog.prod.ListingAlertsEmailOrganizerInstant_alertSetting,1]
> -> 13572)
> 2017-07-25 17:11:29.247 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to
> interrupt.
> 2017-07-25 17:11:29.247 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Shutting down BrokerProxy for hostname:portnum
> 2017-07-25 17:11:29.248 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] closing simple consumer...
> 2017-07-25 17:11:29.265 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to
> interrupt.
> 2017-07-25 17:11:29.265 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Shutting down BrokerProxy for hostname:portnum
> 2017-07-25 17:11:29.265 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] closing simple consumer...
> 2017-07-25 17:11:29.523 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to
> interrupt.
> 2017-07-25 17:11:29.524 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Shutting down BrokerProxy for hostname:portnum
> 2017-07-25 17:11:29.524 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] closing simple consumer...
> 2017-07-25 17:11:29.601 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to
> interrupt.
> 2017-07-25 17:11:29.602 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] Shutting down BrokerProxy for hostname:portnum
> 2017-07-25 17:11:29.602 [main] org.apache.samza.system.kafka.BrokerProxy
> [INFO] closing simple consumer...
> 2017-07-25 17:11:29.663 [SAMZA-BROKER-PROXY-BrokerProxy thread pointed at
> hostname:portnum for client
> samza_consumer-prod_ListingAlertsEmailOrganizerInstant-1]
> org.apache.samza.system.kafka.BrokerProxy [INFO] Shutting down due to
> interrupt.
> 2017-07-25 17:11:29.668 [main] org.apache.samza.container.SamzaContainer
> [INFO] Starting host statistics monitor
> 2017-07-25 17:11:29.670 [main] org.apache.samza.container.SamzaContainer
> [INFO] Registering task instances with producers.
> 2017-07-25 17:11:29.674 [main] org.apache.samza.container.SamzaContainer
> [INFO] Starting producer multiplexer.
> 2017-07-25 17:11:29.675 [main] org.apache.samza.container.SamzaContainer
> [INFO] Initializing stream tasks.
> 2017-07-25 17:11:29.676 [main]
> com.company.samza.app.companyStreamingAppWrapper [INFO] Initializing instance
> of streaming application
> 2017-07-25 17:11:29.681 [main]
> com.company.samza.app.companyStreamingAppWrapper [INFO] First initialization.
> Setting up Guice container with configuration
> companyStreamingAppWrapperConfiguration{company.app.name=ListingAlertsEmailOrganizerInstant,
> company.appgroup=aws, company.env=prod,
> company.guice.module=com.company.notifications.listingAlerts.organizer.email.ListingAlertsEmailOrganizerModule}
> 2017-07-25 17:11:30.118 [main] com.company.config.guice.configModule [INFO]
> configModule loaded requested override file
> '/storage/data/secure/config/AnalyticsServiceClient.cfg'
> 2017-07-25 17:11:30.480 [main]
> com.company.samza.dataService.SamzaSessionFactoriesModule [INFO] Loading prod
> dbConfig from /data/config/prod.database.properties
> // Hibernate stuff (i.e. our code is hit)
> ```
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
