How do you kill off a completed thread?

[tony.herstell1]

How do you kill off a completed thread?

public void getCustomers() {

        this.logger.info(">>>getCustomers");

        // Start a child process and return to the caller.

        State threadState = customerHarvesterThread.getState();

        this.logger.info("Harvesting Thread State:" + threadState.name());

        if (threadState == State.NEW) {

            this.logger.info("Harvesting Thread Starting.");

            customerHarvesterThread.setName("Customer Harvester");

            customerHarvesterThread.start();

            this.logger.info("Harvesting Thread Started:" + customerHarvesterThread.getName() + " (" + customerHarvesterThread.getId() + ")");

        } else {

            if (threadState == State.TERMINATED) {

                // How do we clear the thread as its never purged!

                this.postGlobalMessage("thread_still_engaged", FacesMessage.SEVERITY_INFO);

            } else {

                // Its either running or terminated but not cleaned up yet

                this.postGlobalMessage("thread_still_engaged", FacesMessage.SEVERITY_INFO);

            }

        }

    }

[nickarls]

I thought the thread was going the normal way of GC, references allowing(?)

[tony.herstell1]

Never seems to get cleaned up.

Also If I fire Events in the thread and @Observe them in the parent then the observer never sees them...

[nickarls]

Show more code. How are you firing them? What scope is the observer? How is the thread look etc.

[dlofthouse]

Looking at the posted code so far you appear to have a reference to the Thread for you to be able to check the state - as long as you keep a reference to the thread it will never be garbage collected.

[tony.herstell1]

Observers class (harvest the errors and shows them on a JSF screen)

@URLMappings(mappings = { @URLMapping(id = "mappingErrorsFromImport", pattern = "/customer/errors", viewId = "/pages/customer/importErrors.xhtml")})
@Startup
@Stateful
//Lets be long running (multiple client-server round trips) - Needs Extended on
//PersistanceContext too to hold onto my objects and not get LIEs.
@SessionScoped
//EL Can can find me...
@Named
public class CustomerHarvestFromXXErrorsService extends BaseController implements Serializable {

    private static final long serialVersionUID = 3020101639753884992L;

    private Logger logger = Logger.getLogger(CustomerHarvestFromXXErrorsService.class.getName());

    private Collection<SalesPoint> mappingErrorsFromImport = new ArrayList<SalesPoint>();;

    public CustomerHarvestFromXXErrorsService() {
        super();
    }

    public String init() {
        this.logger.info(">>>init");
        this.logger.info("<<<init");
        return "pretty:mappingErrorsFromImport";
    }

    public void appendError(@Observes CustomerImportFailToMapEvent customerImportFailToMapEvent) {
        this.logger.info(">>>appendError");
        if (mappingErrorsFromImport != null) {
            this.logger.info("Called with " + mappingErrorsFromImport.size() + "Errors");
            this.mappingErrorsFromImport.add((SalesPoint)customerImportFailToMapEvent);
        }
        this.logger.info("<<<appendError");
    }

    public void removeError(@Observes CustomerImportMapEvent customerImportMapEvent) {
        this.logger.info(">>>removeError");
        //TODO
        this.logger.info("<<<removeError");
    }


    public Collection<SalesPoint> getErrors() {
        this.logger.info(">>>getErrors");
        this.logger.info("<<<getErrors");
        return this.mappingErrorsFromImport;
    }

    public void clearErrors(@Observes CustomerImportStartEvent customerImportStartEvent) {
        this.logger.info(">>>clearErrors");
        this.mappingErrorsFromImport = new ArrayList<SalesPoint>();
        this.logger.info("<<<clearErrors");
    }

   public String finished() { 
        this.logger.info(">>>finished");
        this.logger.info("<<<finished");
        return "pretty:home";
   }

}

Controller class (kicksoff the harvest Thread to reap some stuff from some nasty SQL)

@URLMappings(mappings = { @URLMapping(id = "importCustomers", pattern = "/customer/import", viewId = "/pages/customer/importCustomers.xhtml")})
@Stateful
//Lets be long running (multiple client-server round trips) - Needs Extended on
//PersistanceContext too to hold onto my objects and not get LIEs.
@ConversationScoped
//EL Can can find me...
@Named
public class CustomerHarvestFromXXService extends BaseController implements Serializable {

    @SuppressWarnings("unused")
    @URLQueryParameter("cid")
    private String cid;

    // Inject this to make conversation magic happen
    @Inject
    private Conversation conversation;

    @Inject
    CustomerHarvesterThread customerHarvesterThread;


    public CustomerHarvestFromXXService() {
        super();
    }

    /*
     * Hack to pass cid to prettyfaces so it can add it to URL - DONT remove.
     */
    public String getCid() {
        return this.conversation.getId();
    }

    public void setCid(String cid) {
        this.cid = cid;
    }

    // A good place to START a conversation from. ** land here off a menu
    // click...
    public String init() {
        this.logger.info(">>>init");
        if (!this.conversation.isTransient()) {
            this.logger.info("Existing conversation found:" + this.conversation.getId() + " Ending it...");
            this.conversation.end();
        }
        this.logger.info("+++CONVERSATION START");
        this.conversation.begin(); // START THE LONG RUNNING CONVERSATION
        this.logger.info("conversation:" + this.conversation.getId());
        return "pretty:importCustomers";
    }


    public void getCustomers() {
        this.logger.info(">>>getCustomers");
        // Start a child process and return to the caller.
        State threadState = customerHarvesterThread.getState();
        this.logger.info("Harvesting Thread State:" + threadState.name());
        if (threadState == State.NEW) {
            this.logger.info("Harvesting Thread Starting.");
            customerHarvesterThread.setName("Customer Harvester");
            customerHarvesterThread.start();
            this.logger.info("Harvesting Thread Started:" + customerHarvesterThread.getName() + " (" + customerHarvesterThread.getId() + ")");
        } else {
            if (threadState == State.TERMINATED) {
                // How do we clear the thread as its never purged!   <<===== How do we clear it away; stays for ever even if set to null!
                this.postGlobalMessage("thread_still_engaged", FacesMessage.SEVERITY_INFO);
            } else {
                // Its either running or terminated but not cleaned up yet TODO Clean this up!
                this.postGlobalMessage("thread_still_engaged", FacesMessage.SEVERITY_INFO);
            }
        }
    }

   public String finished() { // land on here when finished up to return to
                                          // Main Page (a "done" button).
        this.logger.info(">>>finished");
        this.logger.info("+++CONVERSATION END for conversation:" + this.conversation.getId());
        if (!this.conversation.isTransient()) { // May have been an error (don't
                                                // get trapped on the page).
            this.conversation.end(); // END THE LONG RUNNING CONVERSATION
        }
        this.logger.info("<<<finished");
        return "pretty:home";
   }

}

And the Thread!

NOTE: The Transaction stuff is supposed to start a new TXN per batch but doesn't!!!

@Stateless
@TransactionManagement(TransactionManagementType.BEAN)
public class CustomerHarvesterThread extends Thread {


    private Logger logger = Logger.getLogger(CustomerHarvesterThread.class.getName());


    public CustomerHarvesterThread() {
        super();
    }


    @PersistenceContext(type=PersistenceContextType.EXTENDED)
    private EntityManager em;
    
    // Inject a UserTransaction for manual transaction demarcation.
    @Inject
    private UserTransaction userTransaction;
    
    //@Inject
    //private CustomerHarvestFromXXErrorsService customerHarvestFromXXErrorsService; <<== still trying to get Events working... but will resort to hacking it directly....
    
    
    // Events
    @Inject
    @CustomerImportStartEvent
    Event<Thread> customerImportStartEvent;


    @Inject
    @CustomerImportMapEvent
    Event<Customer> customerImportMapEvent;


    @Inject
    @CustomerImportFailToMapEvent
    Event<SalesPoint> customerImportFailToMapEvent;
    
    @Inject
    @CustomerImportEndEvent
    Event<Thread> customerImportEndEvent;


    //@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
    @Override
    public void run() {
        customerImportStartEvent.fire(this); <<=== Dont see this
        this.ripCustomers();
        logger.info("Completed thread action to process Customers...");
        customerImportEndEvent.fire(this); <<=== Dont see this
    }


    //@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
    private void ripCustomers() {

        boolean logEachCustomer = false;
  
        this.logger.info("> getCustomers");
        
        Collection<SalesPoint> customersFromXX = new ArrayList<SalesPoint>();


        Connection con = null;
        String url = "xxxx";
        String dbName = "xxx";
        String driver = "xxx";
        String userName = "xxx";
        String password = "xxx";
        String query = "SELECT DISTINCT some nasty SQL";
        try {
            Class.forName(driver).newInstance();
            con = DriverManager.getConnection(url + dbName, userName, password);
            Statement st = con.createStatement();
            this.logger.info(query);
            ResultSet rs = st.executeQuery(query);
            if (logEachCustomer) {
                this.logger.info("Results:");
            } 
              ... 

        if (customersFromXX.size() == 0) {
            logger.info("No XX Customers to harvest...");
        } else {
            logger.info("Harvesting " + customersFromXX.size() + " XX Customers...");
        }
        
        // Clear any errors in Error Registry as we are going to add some (TODO need to do this as an append in the future and remove successful ones)
        //customerHarvestFromXXErrorsService.clearErrors();  <<== still trying to get Events working... but will resort to hacking it directly....
        
        // For each customer we now process them in slices.
        int sliceSize = 100;
        int sliceCounter = 0;
        Collection<SalesPoint> customersFromXXInBatch = new ArrayList<SalesPoint>();
        Iterator<SalesPoint> customerIterator = customersFromXX.iterator();
        while (customerIterator.hasNext()) {
            customersFromXXInBatch.add(customerIterator.next());
            sliceCounter++;
            if (sliceCounter == sliceSize) {
                this.processXXCustomerBatch(customersFromXXInBatch);
                customersFromXInBatch.clear();
                sliceCounter = 0;
            }
        }
        if (sliceCounter != 0) { // We have some left in the last batch
            this.processXXCustomerBatch(customersFromXXInBatch);
            customersFromXInBatch.clear();
            sliceCounter = 0;
        }
        logger.info("Finished harvesting XX Customers...");
    }


    // Lets start a new (very clean) transaction for handling each customer batch as we need to store the new "real" customer object
    // against the Territory (Don't want to do all the customers in 1 Trans as timeout will kick in)
    //@TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
    private void processXXCustomerBatch(Collection<SalesPoint> customersFromXXInBatch) {


        try {
        logger.info("Processing Batch...");


        userTransaction.begin(); <<== had hoped this would create a new TXN (per batch) to avoid a timeout - It doesnt!

        logger.info("TXN Started..." + userTransaction.toString());
        
        // If the customer already exists (100% match) then ignore/update
        // TODO


        ... 
       
        //Collection<SalesPoint> errors = new ArrayList<SalesPoint>();  <<== still trying to get Events working... but will resort to hacking it directly....
        for (SalesPoint eachCustomerFromXXInThisBatch : customersFromXXInBatch) {
            if (!eachCustomerFromXXInThisBatch.isUsed()) {
                //errors.add(eachCustomerFromXXInThisBatch);
                logger.info("SalesPoint failed to map: "+ eachCustomerFromXXInThisBatch);
                customerImportFailToMapEvent.fire(eachCustomerFromXXInThisBatch); <<=== Dont see this
            }
        }
        //customerHarvestFromXXErrorsService.appendErrors(errors);  <<== still trying to get Events working... but will resort to hacking it directly....


        userTransaction.commit(); <<== had hoped this would create a new TXN (per batch) to avoid a timeout
        logger.info("TXN Comitted..." + userTransaction.toString());



        ...
    }

   ...
}

[nickarls]

If you do a

@Inject
@CustomerImportStartEvent
Event<Thread> customerImportStartEvent;

you would have

public void lurk(@Observes @CustomerImportStartEvent Thread thread)

right?

Never seen a @Stateless used as a Thread before.

[tony.herstell1]

DOH!

>> Never seen a @Stateless used as a Thread before.

As you can see its been a lot of trial an error...

[tony.herstell1]

DOH!

This compiles too!

@Inject
@CustomerImportStartEvent
Event<Thread> customerImportStartEvent;

you would have

public void lurk(@Observes @CustomerImportStartEvent Thread thread)

public void appendError(@Observes @CustomerImportFailToMapEvent SalesPoint customerImportFailToMapEvent) {

public void removeError(@Observes @CustomerImportMapEvent Customer customerImportMapEvent) {

public void clearErrors(@Observes @CustomerImportStartEvent Thread thread) {

Will set some data up so can try tomorrow at home.

Must sleep; thx....

[nickarls]

Sure, you can fire all the events you want but noone has to listen to them ;-)

The @Stateless Thread is probably why you're not seeing the expected interceptor behaviour...

[sfcoy]

I'm not sure what you're trying to do here, but you can't directly spawn threads from EJBs. The spec specifically disallows it.

If you need to do asynchronous work then have a look at @javax.ejb.Asynchronous.

[sfcoy]

@javax.ejb.Startup can only be applied to @javax.ejb.Singleton beans.

From the look of it, CustomerHarvestFromXXErrorsService should probably look like:

{code:java}@URLMappings(mappings = { @URLMapping(id = "mappingErrorsFromImport", pattern = "/customer/errors", viewId = "/pages/customer/importErrors.xhtml")})
@Startup
@Singleton
//Lets be long running (multiple client-server round trips)
//EL Can can find me...
@Named
public class CustomerHarvestFromXXErrorsService extends BaseController {
...
}
{code}

[nickarls]

It would perhaps be a good idea to output a WARN at deployment time for using @Startup on non-singletons (you see cases once in a while)...

[sfcoy]

I think maybe it should fail to deploy actually, since the behaviour is likely to be incorrect.

[nickarls]

Well, the specification doesn't explicitly forbid it, you are just not expected to see any results if you put it on non-singletons. I think there are many other similar annotations that can be misplaced (e.g. @Inject @SessionScoped) that don't do what might be expected. OTOH, you are free to re-use annotations in your application (e.g. check if they are present and then do something). Although in many cases it will just lead to more confusion.